From d7bf0ba7503860226c56978d3ed4742da6bb481d Mon Sep 17 00:00:00 2001 From: Marius Ciepluch <11855163+norandom@users.noreply.github.com> Date: Wed, 7 Oct 2026 18:23:40 +0200 Subject: [PATCH] unseen: sourcing robustness -- retry transient HTTP/connection errors, shrink the advisory journal A live G0 sourcing run failed on both: a transient connection/decode error aborted the whole enumeration (the bare except raised http_failure on the first blip), and the advisory journal stored full raw advisory JSON and reached 1.77 GB on a near-full host. Fix: the JSON transport now retries URLError/timeout/ConnectionError/IncompleteRead/ JSONDecodeError through the same attempt-4 backoff as HTTP 5xx/429 (fatal only after the cap or on a genuinely non-retryable error); enumerate_advisories journals only the fields resolution/extraction consume via source.reduce_advisory (ghsa id, published/withdrawn, type, CWE ids, package ecosystem/name for OSV, fix-commit links), with journal_version bumped to 2 so an old full-format journal is rejected and resume-skip stays intact. Offline: test_unseen_source/draft/extract 133 passed; ruff + format clean; mypy unchanged (pre-existing verify.py lambda only). Co-Authored-By: Claude Opus 4.8 --- benchmarks/unseen/draft.py | 4 +- benchmarks/unseen/source.py | 48 ++++++++++++++----- tests/test_unseen_draft.py | 86 +++++++++++++++++++++++++++++++++++ tests/test_unseen_source.py | 91 +++++++++++++++++++++++++++++++++++++ 4 files changed, 215 insertions(+), 14 deletions(-) diff --git a/benchmarks/unseen/draft.py b/benchmarks/unseen/draft.py index 6597699..2376fcd 100644 --- a/benchmarks/unseen/draft.py +++ b/benchmarks/unseen/draft.py @@ -219,7 +219,7 @@ def inputs(used, seed, known_empty, clones): "used_set_digest": digest({label: sorted(names) for label, names in used.sources.items()}), "known_empty": sorted(known_empty), "clone_limit_bytes": getattr(clones, "limit_bytes", 500 * 1024 * 1024), - "journal_version": 1, + "journal_version": 2, } # Hardcoded design thresholds and algorithms are covered as well as taxonomy data. for name in ("draft", "source", "extract", "eligibility", "journal"): @@ -235,7 +235,7 @@ def enumerate_advisories(github, journal): if key not in journal.rows["advisories"]: journal.boundary() result = github.advisory_page(page) - journal.put("advisories", key, result) + journal.put("advisories", key, {"rows": [source.reduce_advisory(row) for row in result["rows"]], "done": result["done"]}) result = journal.rows["advisories"][key] records.update((row["ghsa_id"], row) for row in result["rows"]) if result["done"]: diff --git a/benchmarks/unseen/source.py b/benchmarks/unseen/source.py index 08dac91..60c5c5b 100644 --- a/benchmarks/unseen/source.py +++ b/benchmarks/unseen/source.py @@ -61,6 +61,23 @@ def ecosystem(row: dict) -> str | None: return next((e for e in ECOSYSTEMS if e in values), None) +def reduce_advisory(row: dict) -> dict: + """Keep only selection inputs, including every fix needed by the exclusion index. + + Preserve the parser's shape so fresh and resumed pages follow the same path. + Commit parents, licenses and OSV brackets are resolved later into Candidate. + """ + return { + "ghsa_id": row["ghsa_id"], + "published_at": row["published_at"], + "type": row.get("type", "reviewed"), + "withdrawn_at": row.get("withdrawn_at"), + "cwes": [{"cwe_id": c["cwe_id"]} for c in row.get("cwes", [])], + "vulnerabilities": [{"package": {key: v["package"][key] for key in ("ecosystem", "name")}} for v in row.get("vulnerabilities", [])], + "references": [f"https://github.com/{name}/commit/{sha}" for name, sha in sorted(fix_links(row))], + } + + def ordered_advisories(rows: list[dict], seed: int) -> list[dict]: """Stable seeded ordering, post-cutoff first in each ecosystem; round-robin strata.""" rng = random.Random(seed) @@ -166,14 +183,17 @@ class JSONTransport: self.rate_limit = {} def get(self, url: str, headers: dict) -> object: - from urllib.error import HTTPError + from http.client import IncompleteRead + from urllib.error import HTTPError, URLError from urllib.request import Request for attempt in range(5): self.attempts += 1 + retry_headers, limited = {}, False try: with self.opener(Request(url, headers=headers), timeout=60) as response: - self.record_rate(getattr(response, "headers", {})) + retry_headers = getattr(response, "headers", {}) + self.record_rate(retry_headers) return json.load(response) except HTTPError as exc: self.record_rate(exc.headers) @@ -185,18 +205,22 @@ class JSONTransport: transient = exc.code in {500, 502, 503, 504} if attempt == 4 or not (limited or transient): raise ValueError("http_failure") from None - try: - delay = float(exc.headers.get("Retry-After", "0")) - if limited and not delay: - delay = max(60, float(exc.headers.get("X-RateLimit-Reset", "0")) - self.clock() + 1) - delay = max(delay, 2**attempt) - if delay > 3605: - raise ValueError("retry_window") - except (TypeError, ValueError): - raise ValueError("http_retry_failed") from None - self.sleeper(delay) + retry_headers = exc.headers + except (URLError, TimeoutError, ConnectionError, IncompleteRead, json.JSONDecodeError): + if attempt == 4: + raise ValueError("http_failure") from None except Exception: raise ValueError("http_failure") from None + try: + delay = float(retry_headers.get("Retry-After", "0")) + if limited and not delay: + delay = max(60, float(retry_headers.get("X-RateLimit-Reset", "0")) - self.clock() + 1) + delay = max(delay, 2**attempt) + if delay > 3605: + raise ValueError("retry_window") + except (TypeError, ValueError): + raise ValueError("http_retry_failed") from None + self.sleeper(delay) raise ValueError("http_failure") def record_rate(self, headers): diff --git a/tests/test_unseen_draft.py b/tests/test_unseen_draft.py index 5bcddcb..bdadd31 100644 --- a/tests/test_unseen_draft.py +++ b/tests/test_unseen_draft.py @@ -525,3 +525,89 @@ def test_journal_fsync_lock_and_corruption(tmp_path, monkeypatch): handle.write(b'{"id":') with pytest.raises(ValueError, match="journal_incomplete_record"), j.Journal(path).open({"seed": 1}): pass + + +def test_advisory_journal_reduces_nested_blobs_and_resumes_by_page(tmp_path): + import json + + raw = json.loads((Path(__file__).resolve().parents[1] / "benchmarks/unseen/fixtures/advisory.json").read_bytes()) + expected = copy.deepcopy(raw) + raw.update(description="raw advisory blob" * 10000, cvss={"vector_string": "unused"}) + raw["cwes"][0]["name"] = "unused CWE description" + raw["vulnerabilities"][0].update(vulnerable_version_range="unused", first_patched_version={"identifier": "unused"}) + raw["vulnerabilities"][0]["package"]["url"] = "https://example.invalid/unused" + raw["references"] = [{"url": raw["references"][0], "type": "FIX"}, {"url": "https://example.invalid/unused"}] + + class Pages: + calls = [] + + def advisory_page(self, page): + self.calls.append(page) + return {"rows": [raw], "done": True, "unused": "raw page blob"} + + api, used = Pages(), e.UsedSet() + parameters = d.inputs(used, 1, set(used.sources), FakeClones()) + path = tmp_path / "journal" + with d.Journal(path).open(parameters) as journal: + rows = d.enumerate_advisories(api, journal) + assert rows == [expected] + saved = json.loads((path / "advisories.jsonl").read_bytes()) + assert saved["value"] == {"rows": [expected], "done": True} + assert b"unused" not in (path / "advisories.jsonl").read_bytes() + assert b"raw advisory blob" not in (path / "advisories.jsonl").read_bytes() + assert b"raw page blob" not in (path / "advisories.jsonl").read_bytes() + with d.Journal(path).open(parameters) as journal: + resumed = d.enumerate_advisories(api, journal) + assert api.calls == [1] and resumed == rows + + class Bracket(FakeOSV): + def get(self, identifier): + return { + "affected": [ + {"package": {"ecosystem": "PyPI", "name": "fixture"}, "ranges": [{"type": "SEMVER", "events": [{"introduced": "1.2"}]}]} + ] + } + + decision = d.decide(resumed[0], FakeAPI(), Bracket(), used, set()) + candidate = s.Candidate(**decision["candidate"]) + assert candidate.repository == "fixture/project" + assert candidate.fix == "a" * 40 and candidate.parent == "b" * 40 + assert candidate.first_affected == "1.2" and candidate.post_cutoff and candidate.family == "injection" + assert s.candidate(raw, FakeAPI(), Bracket()) == candidate + + +def test_reduced_advisories_preserve_all_fix_exclusions_and_selection(): + rows = list(FakeAPI(2).advisories()) + rows[0].update(published_at="2018-01-01", type="unreviewed", withdrawn_at="2020-01-01", cwes=[]) + rows[0]["references"].append("https://github.com/fixture/project001/commit/" + "c" * 40) + rows[0]["vulnerabilities"].append({"package": {"ecosystem": "rubygems", "name": "other"}}) + reduced = [s.reduce_advisory(row) for row in rows] + assert ( + d.advisory_index(reduced, FakeAPI()) + == d.advisory_index(rows, FakeAPI()) + == {"fixture/project000": {"a" * 40}, "fixture/project001": {"a" * 40, "c" * 40}} + ) + assert [r["ghsa_id"] for r in s.ordered_advisories(rows, 7)] == [r["ghsa_id"] for r in s.ordered_advisories(reduced, 7)] + for key, value, reason in [ + ("published_at", "2018-01-01", "date"), + ("type", "unreviewed", "advisory_status"), + ("withdrawn_at", "2026-01-01", "advisory_status"), + ("cwes", [], "family"), + ]: + raw = {**rows[1], key: value} + with pytest.raises(s.Rejected, match=f"^{reason}$"): + s.candidate(s.reduce_advisory(raw), FakeAPI(), FakeOSV()) + + +def test_old_raw_advisory_journal_version_refused_before_calls(tmp_path): + used, api, osv, clones = e.UsedSet(), FakeAPI(1), FakeOSV(), FakeClones() + path = tmp_path / "journal" + parameters = d.inputs(used, 7, set(used.sources), clones) + assert parameters["journal_version"] == 2 + with d.Journal(path).open({**parameters, "journal_version": 1}) as journal: + journal.put("advisories", "1", {"rows": list(api.advisories()), "done": True}) + report = d.run( + tmp_path / "repo", github=api, osv=osv, clones=clones, used=used, seed=7, known_empty=set(used.sources), journal_path=path + ) + assert report["status"] == "resume_refused" and report["changed_parameters"] == ["journal_version"] + assert api.calls == osv.calls == 0 and clones.counts == {} diff --git a/tests/test_unseen_source.py b/tests/test_unseen_source.py index 8d9d587..37804b5 100644 --- a/tests/test_unseen_source.py +++ b/tests/test_unseen_source.py @@ -180,3 +180,94 @@ def test_transport_rate_limit_retry_and_sanitized_failure(): with pytest.raises(ValueError, match="^http_failure$"): transport.get("https://example.invalid", {}) assert transport.attempts == 5 + + +@pytest.mark.parametrize("failures", [1, 3, 4, 5, 7]) +@pytest.mark.parametrize("kind", ["url", "timeout", "connection", "reset", "incomplete", "json"]) +def test_transport_connection_and_body_transients(kind, failures): + import io + from http.client import IncompleteRead + from urllib.error import URLError + + errors = { + "url": URLError("private endpoint"), + "timeout": TimeoutError("private endpoint"), + "connection": ConnectionError("private endpoint"), + "reset": ConnectionResetError("private endpoint"), + "incomplete": IncompleteRead(b"private partial body", 100), + } + waits, calls = [], [] + + class BrokenBody(io.BytesIO): + def read(self, *args): + raise errors[kind] + + def opener(request, timeout): + calls.append(request) + if len(calls) <= failures: + if kind == "json": + return io.BytesIO(b'{"private":') # exercise json.load, not just the opener + if kind == "incomplete": + return BrokenBody() + raise errors[kind] + return io.BytesIO(b'{"ok": true}') + + transport = s.JSONTransport(opener=opener, sleeper=waits.append) + if failures >= 5: + with pytest.raises(ValueError, match="^http_failure$"): + transport.get("https://example.invalid", {}) + else: + assert transport.get("https://example.invalid", {}) == {"ok": True} + assert transport.attempts == len(calls) == min(failures + 1, 5) + assert waits == [2**i for i in range(min(failures, 4))] + + +@pytest.mark.parametrize("code", [400, 401, 403, 404, 422]) +def test_transport_nonretryable_http_status(code): + from urllib.error import HTTPError + + waits = [] + + def opener(request, timeout): + raise HTTPError(request.full_url, code, "private endpoint", {}, None) + + transport = s.JSONTransport(opener=opener, sleeper=waits.append) + error, message = (s.RepositoryNotFound, "not_found") if code == 404 else (ValueError, "http_failure") + with pytest.raises(error, match=f"^{message}$"): + transport.get("https://example.invalid", {}) + assert transport.attempts == 1 and waits == [] + + +@pytest.mark.parametrize("code", [403, 429, 500, 502, 503, 504]) +@pytest.mark.parametrize("retry_after", ["7", "invalid", "3606"]) +def test_transport_http_retry_after_and_window_guard(code, retry_after): + import io + from urllib.error import HTTPError + + waits, calls = [], [] + + def opener(request, timeout): + calls.append(request) + if len(calls) == 1: + raise HTTPError(request.full_url, code, "private endpoint", {"Retry-After": retry_after}, None) + return io.BytesIO(b"{}") + + transport = s.JSONTransport(opener=opener, sleeper=waits.append) + if retry_after == "7": + assert transport.get("https://example.invalid", {}) == {} + assert waits == [7] and transport.attempts == 2 + else: + with pytest.raises(ValueError, match="^http_retry_failed$"): + transport.get("https://example.invalid", {}) + assert waits == [] and transport.attempts == 1 + + +def test_transport_decode_retry_respects_response_retry_after(): + import io + + waits = [] + responses = [io.BytesIO(b'{"truncated":'), io.BytesIO(b"{}")] + responses[0].headers = {"Retry-After": "9"} + transport = s.JSONTransport(opener=lambda *a, **kw: responses.pop(0), sleeper=waits.append) + assert transport.get("https://example.invalid", {}) == {} + assert waits == [9]