From 06b8a40c46a309a9c3a00d3514849904b6980c97 Mon Sep 17 00:00:00 2001 From: Marius Ciepluch <11855163+norandom@users.noreply.github.com> Date: Sat, 10 Oct 2026 16:33:29 +0200 Subject: [PATCH] unseen: cache enumeration + resolution to ousast-corpus (restarts skip the ~26-min prefix) The decision bank made extraction durable, but every restart still re-paid the ~26-minute advisory enumeration + per-advisory resolution, since those lived only in the local journal. Mirror them into the bucket so a restart reads them back and goes essentially straight to any remaining extraction. bank.py: bank_resolution/resolved at //resolved/.json (streamed one-by-one as each advisory is decided), cache_advisories/ cached_advisories at //advisories/list.json, and bank_repository/repositories (caches the advisory_index repo-resolution pass, preserving renames and 404s). All gated by RESOLUTION_CONTRACT="v1" (extract.py) -- bumped only when enumeration/eligibility/source.candidate output changes; recent fetch/parse/pool changes do not bump it. draft.py run(): with --bank, advisories load from the cache (else enumerate then cache); eligibility decisions load from the bucket (contract-matched) to skip re-resolution; new resolutions stream to the bucket before the local checkpoint write. Repository lookups reuse the repositories/ cache. No --bank => local-journal behavior unchanged. Net: advisories + resolutions + extraction all resume from ousast-corpus; restarts no longer re-enumerate or re-resolve. Fresh and resumed runs produce byte-identical manifests at 1 and 4 workers. Bank methods proven live against ousast-corpus. 356 tests pass, ruff clean. Co-Authored-By: Claude Opus 4.8 --- benchmarks/unseen/bank.py | 86 ++++++++++++++ benchmarks/unseen/draft.py | 84 +++++++++++--- benchmarks/unseen/extract.py | 2 + tests/test_unseen_bank.py | 101 ++++++++++++++++ tests/test_unseen_draft.py | 219 ++++++++++++++++++++++++++++++++++- 5 files changed, 477 insertions(+), 15 deletions(-) diff --git a/benchmarks/unseen/bank.py b/benchmarks/unseen/bank.py index ec9cc6a..73e9533 100644 --- a/benchmarks/unseen/bank.py +++ b/benchmarks/unseen/bank.py @@ -105,6 +105,92 @@ class CorpusBank: self.client = client self.seed = seed self.decisions_prefix = f"{prefix}/{seed}/decisions/" + self.resolved_prefix = f"{prefix}/{seed}/resolved/" + self.advisories_key = f"{prefix}/{seed}/advisories/list.json" + self.repositories_prefix = f"{prefix}/{seed}/repositories/" + + @staticmethod + def advisory_key(advisory_id: str) -> str: + return hashlib.sha256(advisory_id.encode()).hexdigest() + + def _put(self, key: str, body: dict, *, labels: dict) -> None: + existing = _retry(self.client.get, key) + if existing is not None: + previous = json.loads(existing[0]) + if {field: value for field, value in previous.items() if field != "ts"} == body: + return + body = {**body, "ts": datetime.now(UTC).isoformat()} + data = json.dumps(body, sort_keys=True, separators=(",", ":"), ensure_ascii=False).encode() + _retry(self.client.put, key, data, labels=labels) + + def bank_resolution(self, advisory_id: str, result: dict, *, contract: str) -> None: + decision = "candidate" if "candidate" in result else "reject" + self._put( + self.resolved_prefix + self.advisory_key(advisory_id) + ".json", + { + "advisory": advisory_id, + "decision": decision, + "result": result, + "reason": result.get("rejected"), + "resolution_contract": contract, + "seed": self.seed, + }, + labels={"decision": decision, "contract": contract}, + ) + + def _listed(self, prefix: str, identity: str) -> dict[str, dict]: + results = {} + for key in sorted(self.client.keys(prefix)): + stored = _retry(self.client.get, key) + if stored is None: + raise ValueError("corpus_resolution_missing") + body = json.loads(stored[0]) + name = body[identity] + if key != prefix + self.advisory_key(name) + ".json" or body["seed"] != self.seed: + raise ValueError("corpus_resolution_identity") + results[name] = body + return results + + def resolved(self) -> dict[str, dict]: + results = self._listed(self.resolved_prefix, "advisory") + for body in results.values(): + expected = "candidate" if "candidate" in body["result"] else "reject" + if body["decision"] != expected: + raise ValueError("corpus_resolution_invalid") + return results + + def cache_advisories(self, rows: list, *, contract: str) -> None: + if self.cached_advisories(contract=contract) is not None: + return + self._put( + self.advisories_key, + {"advisories": rows, "advisories_contract": contract, "seed": self.seed}, + labels={"contract": contract}, + ) + + def cached_advisories(self, *, contract: str) -> list | None: + stored = _retry(self.client.get, self.advisories_key) + if stored is None: + return None + body = json.loads(stored[0]) + if body["seed"] != self.seed: + raise ValueError("corpus_advisories_identity") + return body["advisories"] if body["advisories_contract"] == contract else None + + def bank_repository(self, name: str, result: dict | None, *, contract: str) -> None: + """Retain redirects, fork identity and 404s needed by the all-fix index.""" + self._put( + self.repositories_prefix + self.advisory_key(name) + ".json", + {"repository": name, "result": result, "resolution_contract": contract, "seed": self.seed}, + labels={"contract": contract}, + ) + + def repositories(self, *, contract: str) -> dict[str, dict | None]: + return { + name: body["result"] + for name, body in self._listed(self.repositories_prefix, "repository").items() + if body["resolution_contract"] == contract + } @staticmethod def key_for(repository: str, fix: str) -> str: diff --git a/benchmarks/unseen/draft.py b/benchmarks/unseen/draft.py index d90ea09..da52ba4 100644 --- a/benchmarks/unseen/draft.py +++ b/benchmarks/unseen/draft.py @@ -26,7 +26,7 @@ from pathlib import Path from . import eligibility, extract, source from .bank import CorpusBank, CorpusObjects, corpus_settings, split_decided -from .extract import EXTRACTION_CONTRACT, Git, InstrumentFailure +from .extract import EXTRACTION_CONTRACT, RESOLUTION_CONTRACT, Git, InstrumentFailure from .journal import Journal, ResolvedGitHub, ResumeMismatch, Stopped, atomic, digest, external @@ -209,6 +209,43 @@ def prove_source(git, revision: str) -> dict: raise InstrumentFailure("zero_source_bytes") +class BankedGitHub(ResolvedGitHub): + """Stream the fix index's repository metadata before continuing the draw.""" + + def __init__(self, github, journal, bank): + super().__init__(github, journal) + self.bank = bank + self.resolutions = bank.repositories(contract=RESOLUTION_CONTRACT) + journal.rows["resolution"].update({name: body["metadata"] if body is not None else None for name, body in self.resolutions.items()}) + + def resolve(self, name): + key = eligibility.normalize(name) + if key not in self.resolutions: + try: + resolved = eligibility.resolve(name, self) + except eligibility.RepositoryNotFound: + body = None + else: + body = { + "metadata": self.journal.rows["resolution"].get(key), + "full_name": resolved.full_name, + "names": sorted(resolved.names), + "network": resolved.network, + } + self.bank.bank_repository(key, body, contract=RESOLUTION_CONTRACT) + self.resolutions[key] = body + body = self.resolutions[key] + if body is None: + raise eligibility.RepositoryNotFound("not_found") + return eligibility.Resolution(body["full_name"], frozenset(body["names"]), body["network"]) + + +def resolve_repository(name, github): + if isinstance(github, BankedGitHub): + return github.resolve(name) + return eligibility.resolve(name, github) + + def advisory_index(rows: list[dict], github) -> dict[str, set[str]]: """Include ALL fixes in the enumerated population, even from rejected candidates. @@ -219,7 +256,7 @@ def advisory_index(rows: list[dict], github) -> dict[str, set[str]]: for row in rows: for name, sha in source.fix_links(row): try: - resolved = eligibility.resolve(name, github) + resolved = resolve_repository(name, github) except eligibility.RepositoryNotFound: index[name].add(sha) continue @@ -305,7 +342,7 @@ def decide(row, github, osv, used, selected_networks=()): if len(links) != 1: raise source.Rejected("fix_links") name, _ = next(iter(links)) - resolved = eligibility.resolve(name, github) + resolved = resolve_repository(name, github) rejected_by = eligibility.rejection(resolved, used, set()) if rejected_by: raise source.Rejected("used_" + rejected_by) @@ -382,6 +419,7 @@ def ordered_extractions( pool_size, bank: CorpusBank | None = None, decided=None, + stale_resolutions=(), ): """Bounded in-flight work, completion journaling, and seed-order acceptance. @@ -400,6 +438,15 @@ def ordered_extractions( context = multiprocessing.get_context("fork") pool = ProcessPoolExecutor(max_workers=workers, mp_context=context) + def resolution(row): + key = row["ghsa_id"] + if key not in checkpoint.rows["eligibility"]: + result = decide(row, api, osv, used) + if bank is not None: + bank.bank_resolution(key, result, contract=RESOLUTION_CONTRACT) + checkpoint.put("eligibility", key, result) + return checkpoint.rows["eligibility"][key] + def remember(repository, fix, result): bank.bank(repository, fix, result, contract=EXTRACTION_CONTRACT) # Reuse decisions made during this invocation too: another advisory for @@ -461,13 +508,15 @@ def ordered_extractions( exhausted = True break key = row["ghsa_id"] + if key in stale_resolutions: + resolution(row) identity = None if bank is not None: links = source.fix_links(row) if len(links) == 1: repository, fix = next(iter(links)) try: - resolved = eligibility.resolve(repository, api) + resolved = resolve_repository(repository, api) repository = resolved.full_name except eligibility.RepositoryNotFound: resolved = None @@ -508,9 +557,7 @@ def ordered_extractions( checkpoint.put("extraction", extraction_key, result) pending.append((row, decision, extraction_key)) break - if key not in checkpoint.rows["eligibility"]: - checkpoint.put("eligibility", key, decide(row, api, osv, used)) - decision = checkpoint.rows["eligibility"][key] + decision = resolution(row) extraction_key, future = None, None if "candidate" in decision: candidate = source.Candidate(**decision["candidate"]) @@ -619,6 +666,7 @@ def run( scratch = ExitStack() try: decided = bank.decided() if bank is not None else {} + banked = bank.resolved() if bank is not None else {} accepted, reusable = split_decided(decided, contract=EXTRACTION_CONTRACT) decided = {key: value for key, value in decided.items() if key in reusable} known_empty = set(known_empty or ()) @@ -638,16 +686,23 @@ def run( ) with checkpoint.open(inputs(used, seed, known_empty, clones, pool_size=pool_size, slices=slices)): checkpoint.boundary() - api = ResolvedGitHub(github, checkpoint) + api = BankedGitHub(github, checkpoint, bank) if bank is not None else ResolvedGitHub(github, checkpoint) + checkpoint.rows["eligibility"].update( + {key: body["result"] for key, body in banked.items() if body["resolution_contract"] == RESOLUTION_CONTRACT} + ) used = copy.deepcopy(used) # canonicalization must not mutate the resume input eligibility.canonicalize_used(used, api) report["used_set"] = used.report(known_empty) - try: - rows = enumerate_advisories(api, checkpoint) - except InstrumentFailure as exc: - if str(exc) != "empty_advisories" or not accepted: - raise - rows = [] + rows = bank.cached_advisories(contract=RESOLUTION_CONTRACT) if bank is not None else None + if rows is None: + try: + rows = enumerate_advisories(api, checkpoint) + except InstrumentFailure as exc: + if str(exc) != "empty_advisories" or not accepted: + raise + rows = [] + if bank is not None: + bank.cache_advisories(rows, contract=RESOLUTION_CONTRACT) if bank is not None: # Retain accepted entries even if their advisory disappeared upstream. present = {row["ghsa_id"] for row in rows} @@ -680,6 +735,7 @@ def run( pool_size, bank=bank, decided=decided, + stale_resolutions={key for key, body in banked.items() if body["resolution_contract"] != RESOLUTION_CONTRACT}, ) as outcomes: for row, decision, result in outcomes: counts[source.ecosystem(row)] += 1 diff --git a/benchmarks/unseen/extract.py b/benchmarks/unseen/extract.py index 7e16449..dc6d9f2 100644 --- a/benchmarks/unseen/extract.py +++ b/benchmarks/unseen/extract.py @@ -21,6 +21,8 @@ from .source import Candidate, Rejected # Bump only when accept/reject outcomes or accepted entry bytes change. EXTRACTION_CONTRACT = "v1" +# Bump only when enumeration, eligibility or source.candidate outputs change. +RESOLUTION_CONTRACT = "v1" LANGUAGES = {".py", ".js", ".jsx", ".mjs", ".cjs", ".ts", ".tsx", ".php", ".java"} HUNK = re.compile(r"^@@ -(\d+)(?:,(\d+))? \+(\d+)(?:,(\d+))? @@", re.M) diff --git a/tests/test_unseen_bank.py b/tests/test_unseen_bank.py index 13f9f02..8d4fae9 100644 --- a/tests/test_unseen_bank.py +++ b/tests/test_unseen_bank.py @@ -214,3 +214,104 @@ def test_import_does_not_require_boto3(monkeypatch): module = importlib.util.module_from_spec(spec) spec.loader.exec_module(module) assert module.CorpusBank.key_for("a", "b") == CorpusBank.key_for("a", "b") + + +@pytest.mark.parametrize( + "result", + [ + {"candidate": {"repository": "org/é"}, "network": "org/root"}, + {"rejected": "used_pairs"}, + {"rejected": "family", "network": "org/root"}, + ], +) +def test_resolution_roundtrip_idempotent_and_contract_replacement(result): + objects = FakeObjects() + bank = CorpusBank(objects, seed=42, prefix="custom") + advisory = "GHSA-test/advisory" + key = f"custom/42/resolved/{hashlib.sha256(advisory.encode()).hexdigest()}.json" + assert bank.advisory_key(advisory) == hashlib.sha256(advisory.encode()).hexdigest() + bank.bank_resolution(advisory, result, contract="old") + first = objects.objects.copy() + bank.bank_resolution(advisory, result, contract="old") + assert objects.objects == first and objects.calls.count("put") == 1 + row = bank.resolved()[advisory] + assert row == { + "advisory": advisory, + "result": result, + "decision": "candidate" if "candidate" in result else "reject", + "reason": result.get("rejected"), + "resolution_contract": "old", + "seed": 42, + "ts": row["ts"], + } + assert list(objects.objects) == [key] + assert datetime.fromisoformat(row["ts"]).utcoffset().total_seconds() == 0 + bank.bank_resolution(advisory, result, contract="v1") + assert bank.resolved()[advisory]["resolution_contract"] == "v1" + assert objects.calls.count("put") == 2 + + +def test_advisory_snapshot_written_once_per_contract(): + objects = FakeObjects() + bank = CorpusBank(objects, seed=7) + rows = [{"ghsa_id": "GHSA-é", "references": ["a", "b"]}] + assert bank.cached_advisories(contract="v1") is None + bank.cache_advisories(rows, contract="v1") + original = objects.objects.copy() + assert bank.cached_advisories(contract="v1") == rows + assert bank.cached_advisories(contract="v2") is None + bank.cache_advisories([], contract="v1") + assert objects.objects == original and objects.calls.count("put") == 1 + body = json.loads(objects.objects["unseen/7/advisories/list.json"]) + assert body == {"advisories": rows, "advisories_contract": "v1", "seed": 7, "ts": body["ts"]} + bank.cache_advisories([], contract="v2") + assert bank.cached_advisories(contract="v2") == [] + assert bank.cached_advisories(contract="v1") is None + + +@pytest.mark.parametrize("method", ["resolution", "advisories", "repository"]) +@pytest.mark.parametrize("operation", ["get", "put"]) +@pytest.mark.parametrize("failures", [1, 2]) +def test_new_bank_writes_retry_once(method, operation, failures): + objects = FakeObjects() + bank = CorpusBank(objects, seed=7) + objects.failures[operation] = failures + + def write(): + if method == "resolution": + bank.bank_resolution("GHSA-a", {"rejected": "family"}, contract="v1") + elif method == "advisories": + bank.cache_advisories([], contract="v1") + else: + bank.bank_repository("org/repo", None, contract="v1") + + if failures == 2: + with pytest.raises(OSError): + write() + assert objects.calls.count(operation) == 2 + else: + write() + assert len(objects.objects) == 1 + assert objects.failures[operation] == 0 + + +@pytest.mark.parametrize("method", ["resolved", "cached_advisories", "repositories"]) +@pytest.mark.parametrize("failures", [1, 2]) +def test_new_bank_reads_retry_once(method, failures): + objects = FakeObjects() + bank = CorpusBank(objects, seed=7) + bank.bank_resolution("GHSA-a", {"rejected": "family"}, contract="v1") + bank.cache_advisories([], contract="v1") + bank.bank_repository("org/old", {"full_name": "org/new"}, contract="v1") + bank.bank_repository("org/missing", None, contract="old") + objects.failures["get"] = failures + read = getattr(bank, method) + kwargs = {} if method == "resolved" else {"contract": "v1"} + if failures == 2: + with pytest.raises(OSError): + read(**kwargs) + else: + result = read(**kwargs) + assert result is not None + if method == "repositories": + assert result == {"org/old": {"full_name": "org/new"}} diff --git a/tests/test_unseen_draft.py b/tests/test_unseen_draft.py index 975d9fd..2f3f7e2 100644 --- a/tests/test_unseen_draft.py +++ b/tests/test_unseen_draft.py @@ -1651,6 +1651,8 @@ def test_bank_crash_resume_and_durability_failure(tmp_path, monkeypatch, failure puts = 0 def put(self, key, data, labels): + if "/decisions/" not in key: + return super().put(key, data, labels) self.puts += 1 if failure == "persistent_put" and not self.interrupted: raise OSError("offline simulated bank outage") @@ -1687,7 +1689,7 @@ def test_bank_crash_resume_and_durability_failure(tmp_path, monkeypatch, failure report = run(tmp_path / "resume", bank) assert report["status"] == "instrument_failure" assert objects.puts == 2 - assert not objects.objects + assert not bank.decided() assert not (tmp_path / "resume" / "draft-p1.toml").exists() objects.interrupted = True resumed = run(tmp_path / "resume", bank) @@ -1817,3 +1819,218 @@ def test_bank_duplicate_advisory_cannot_overwrite_pair_decision(tmp_path, monkey assert all(report["status"] == "draft" for report in reports) assert reports[0]["manifest_sha256"] == reports[1]["manifest_sha256"] assert bank.decided()[bank.key_for(repository, fix)]["decision"] == "accept" + + +@pytest.mark.parametrize("workers", [1, 4]) +def test_bank_enumeration_resolution_resume_and_contracts(tmp_path, monkeypatch, workers): + from benchmarks.unseen.bank import CorpusBank + + monkeypatch.setattr(importlib.import_module("benchmarks.unseen.journal").os, "fsync", lambda fd: None) + rows = s.ordered_advisories(list(FakeAPI(8).advisories()), 7) + rows[0]["cwes"] = [] # Bank a rejected advisory as well as candidates. + names = [next(iter(s.fix_links(row)))[0] for row in rows] + changes = entries()[0]["changes"] + monkeypatch.setattr(d, "repository_changes", lambda *a, **kw: (copy.deepcopy(changes), {}, {"bytes": 12, "files": 1})) + calls = Counter() + original_enumerate, original_decide, original_candidate = d.enumerate_advisories, d.decide, s.candidate + original_resolve = e.resolve + + def resolve(*args): + calls["resolve"] += 1 + return original_resolve(*args) + + monkeypatch.setattr(e, "resolve", resolve) + + class API(FakeAPI): + def advisories(self): + return iter(copy.deepcopy(rows)) + + def get_repo(self, name): + calls["get_repo"] += 1 + result = super().get_repo(name) + if name == names[2]: + result["source"] = {"full_name": names[1]} + return result + + def enumerate_rows(*args): + calls["enumerate"] += 1 + return original_enumerate(*args) + + def decide(row, *args): + calls[row["ghsa_id"]] += 1 + return original_decide(row, *args) + + def candidate(*args): + calls["candidate"] += 1 + return original_candidate(*args) + + monkeypatch.setattr(d, "enumerate_advisories", enumerate_rows) + monkeypatch.setattr(d, "decide", decide) + monkeypatch.setattr(s, "candidate", candidate) + + class Objects(CorpusFakeObjects): + def __init__(self): + super().__init__() + self.puts = Counter() + + def put(self, key, data, labels): + self.puts[key] += 1 + super().put(key, data, labels) + + objects = Objects() + bank = CorpusBank(objects, seed=7) + + def run(name, bank): + used = e.UsedSet() + return d.run( + tmp_path / "repo", + github=API(), + osv=FakeOSV(), + clones=FakeClones(), + used=used, + seed=7, + known_empty=set(used.sources), + journal_path=tmp_path / name, + extract_workers=workers, + pool_size=6, + slices=1, + bank=bank, + ) + + full = run("full", None) + assert full["status"] == "draft", full + calls.clear() + first = run("first", bank) + assert first["manifest_sha256"] == full["manifest_sha256"] + assert calls["enumerate"] == 1 and calls["candidate"] == 8 + resolved = bank.resolved() + assert len(resolved) == 8 + assert {body["decision"] for body in resolved.values()} == {"candidate", "reject"} + assert all(objects.puts[key] == 1 for key in objects.objects if "/resolved/" in key or "/advisories/" in key) + assert all(body["result"] == original_decide(row, API(), FakeOSV(), e.UsedSet()) for row in rows for body in [resolved[row["ghsa_id"]]]) + calls.clear() + second = run("second", bank) + assert not calls # No enumeration, decide, candidate, or upstream repository lookup. + assert second["manifest_sha256"] == first["manifest_sha256"] + for filename in ("draft-p1.toml", "private-draft-p1.toml"): + assert (tmp_path / "second" / filename).read_bytes() == (tmp_path / "full" / filename).read_bytes() + + # A restart with resolution complete but extraction unfinished also uses + # the exact cached candidates, without re-running any prefix operation. + for key in list(objects.objects): + if "/decisions/" in key: + del objects.objects[key] + remaining = run("remaining-extraction", bank) + assert remaining["manifest_sha256"] == full["manifest_sha256"] + assert not calls + + stale_id = rows[1]["ghsa_id"] + bank.bank_resolution(stale_id, {"rejected": "old"}, contract="old") + stale = run("stale-resolution", bank) + assert stale["manifest_sha256"] == full["manifest_sha256"] + assert calls == {stale_id: 1, "candidate": 1} + assert bank.resolved()[stale_id]["resolution_contract"] == d.RESOLUTION_CONTRACT + bank.cache_advisories([{"ghsa_id": "obsolete"}], contract="old") + calls.clear() + renewed = run("stale-advisories", bank) + assert renewed["manifest_sha256"] == full["manifest_sha256"] + assert calls == {"enumerate": 1} + assert s.ordered_advisories(bank.cached_advisories(contract=d.RESOLUTION_CONTRACT), 7) == s.ordered_advisories( + [s.reduce_advisory(row) for row in rows], 7 + ) + + +@pytest.mark.parametrize("stage", ["advisories", "repositories", "resolved"]) +@pytest.mark.parametrize("failure", ["interrupt", "put_failure"]) +def test_prefix_streaming_failure_and_restart(tmp_path, monkeypatch, stage, failure): + from benchmarks.unseen.bank import CorpusBank + + monkeypatch.setattr(importlib.import_module("benchmarks.unseen.journal").os, "fsync", lambda fd: None) + changes = entries()[0]["changes"] + monkeypatch.setattr(d, "repository_changes", lambda *a, **kw: (copy.deepcopy(changes), {}, {"bytes": 1, "files": 1})) + calls = Counter() + original = d.decide + + def decide(row, *args): + calls[row["ghsa_id"]] += 1 + return original(row, *args) + + monkeypatch.setattr(d, "decide", decide) + + class Objects(CorpusFakeObjects): + armed = True + attempts = 0 + + def put(self, key, data, labels): + if self.armed and f"/{stage}/" in key: + self.attempts += 1 + if failure == "put_failure": + raise OSError("offline outage") + super().put(key, data, labels) + raise KeyboardInterrupt + super().put(key, data, labels) + + objects = Objects() + bank = CorpusBank(objects, seed=7) + + def run(name, bank): + used = e.UsedSet() + return d.run( + tmp_path / "repo", + github=FakeAPI(4), + osv=FakeOSV(), + clones=FakeClones(), + used=used, + seed=7, + known_empty=set(used.sources), + journal_path=tmp_path / name, + pool_size=4, + slices=1, + extract_workers=1, + bank=bank, + ) + + if failure == "interrupt": + with pytest.raises(KeyboardInterrupt): + run("interrupted", bank) + assert objects.attempts == 1 + else: + assert run("interrupted", bank)["status"] == "instrument_failure" + assert objects.attempts == 2 + saved = bank.resolved() + objects.armed = False + calls.clear() + resumed = run("resumed", bank) + assert resumed["status"] == "draft", resumed + assert all(calls[key] == 0 for key in saved) + assert len(bank.resolved()) == 4 + baseline = run("baseline", None) + assert resumed["manifest_sha256"] == baseline["manifest_sha256"] + + +def test_banked_fix_index_preserves_renames_multiple_links_and_404s(tmp_path, monkeypatch): + from benchmarks.unseen.bank import CorpusBank + + rows = list(FakeAPI(3).advisories()) + rows[0]["references"] = ["https://github.com/fixture/old/commit/" + "a" * 40] + rows[1]["references"] = ["https://github.com/fixture/missing/commit/" + "b" * 40] + rows[2]["references"] += ["https://github.com/fixture/old/commit/" + "c" * 40] + + class API(FakeAPI): + def get_repo(self, name): + if name == "fixture/missing": + raise e.RepositoryNotFound("not_found") + return super().get_repo(name) + + bank = CorpusBank(CorpusFakeObjects(), seed=7) + with d.Journal(tmp_path / "fresh").open({}) as checkpoint: + fresh = d.advisory_index(rows, d.BankedGitHub(API(), checkpoint, bank)) + assert fresh == {"fixture/project000": {"a" * 40, "c" * 40}, "fixture/missing": {"b" * 40}, "fixture/project002": {"a" * 40}} + + def unexpected(*args): + pytest.fail("banked fix index must not repeat resolution") + + monkeypatch.setattr(e, "resolve", unexpected) + monkeypatch.setattr(API, "get_repo", unexpected) + with d.Journal(tmp_path / "resumed").open({}) as checkpoint: + assert d.advisory_index(rows, d.BankedGitHub(API(), checkpoint, bank)) == fresh