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 <[email protected]>

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"]:

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:
+                retry_headers = exc.headers
-                    delay = float(exc.headers.get("Retry-After", "0"))
+            except (URLError, TimeoutError, ConnectionError, IncompleteRead, json.JSONDecodeError):
-                    if limited and not delay:
+                if attempt == 4:
-                        delay = max(60, float(exc.headers.get("X-RateLimit-Reset", "0")) - self.clock() + 1)
+                    raise ValueError("http_failure") from None
-                    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)
             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):

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 == {}

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]