unseen: cursor pagination for advisory enumeration

The GitHub /advisories endpoint ignores the `page` query parameter: it
paginates with the Link header's rel="next" cursor. The page-number
pager re-fetched the same first 100 rows forever and never terminated,
so enumeration never completed and no draft converged (a run fetched
5137 "pip pages" holding only 100 unique advisories).

Surface the successful response headers from JSONTransport, and walk the
rel="next" cursor per ecosystem until it is absent. The ecosystem /
published>=2019 / bare-number cwes filters are preserved on the first
request. Journal keys stay ecosystem:n (now cursor-walk positions) with
the saved next URL for resume; journal_version 4 refuses older journals.

Live smoke via the production pager: 300 unique advisories pulled in 3
API calls, advancing past the first page. Offline: 170 unseen tests
pass, ruff clean.

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": 3,
+        "journal_version": 4,
     }
     # Hardcoded design thresholds and algorithms are covered as well as taxonomy data.
     for name in ("draft", "source", "extract", "eligibility", "journal"):
@@ -232,15 +232,19 @@ def enumerate_advisories(github, journal):
     records = {}
     for ecosystem in source.ECOSYSTEMS:
         page = 1
+        next_url = None
         while True:
             key = f"{ecosystem}:{page}"
             if key not in journal.rows["advisories"]:
                 journal.boundary()
-                result = github.advisory_page(page, ecosystem=ecosystem)
-                journal.put("advisories", key, {"rows": [source.reduce_advisory(row) for row in result["rows"]], "done": result["done"]})
+                result = github.advisory_page(next_url, ecosystem=ecosystem)
+                journal.put(
+                    "advisories", key, {"rows": [source.reduce_advisory(row) for row in result["rows"]], "next_url": result["next_url"]}
+                )
             result = journal.rows["advisories"][key]
             records.update((row["ghsa_id"], row) for row in result["rows"])
-            if result["done"]:
+            next_url = result["next_url"]
+            if next_url is None:
                 break
             page += 1
     if not records:

benchmarks/unseen/source.py

@@ -181,6 +181,7 @@ class JSONTransport:
         self.clock = clock or time.time
         self.attempts = 0
         self.rate_limit = {}
+        self.response_headers = {}
 
     def get(self, url: str, headers: dict) -> object:
         from http.client import IncompleteRead
@@ -194,7 +195,9 @@ class JSONTransport:
                 with self.opener(Request(url, headers=headers), timeout=60) as response:
                     retry_headers = getattr(response, "headers", {})
                     self.record_rate(retry_headers)
-                    return json.load(response)
+                    body = json.load(response)
+                    self.response_headers = retry_headers
+                    return body
             except HTTPError as exc:
                 self.record_rate(exc.headers)
                 if exc.code == 404:
@@ -247,7 +250,8 @@ class GitHub:
 
     def get(self, path: str):
         self.calls += 1
-        return self.transport.get("https://api.github.com" + path, self.headers)
+        url = path if path.startswith("https://api.github.com/") else "https://api.github.com" + path
+        return self.transport.get(url, self.headers)
 
     def get_repo(self, name: str) -> dict:
         name = normalize(name)
@@ -269,14 +273,13 @@ class GitHub:
             "text": base64.b64decode(row.get("content", "")).decode("utf-8", errors="replace"),
         }
 
-    def advisory_page(self, page: int, *, ecosystem: str) -> dict:
+    def advisory_page(self, next_url: str | None = None, *, ecosystem: str) -> dict:
         if ecosystem not in ECOSYSTEMS:
             raise ValueError("invalid_ecosystem")
         # Oldest first: newly published advisories do not shift already persisted pages.
         parameters = {
             "type": "reviewed",
             "per_page": 100,
-            "page": page,
             "sort": "published",
             "direction": "asc",
             "ecosystem": ecosystem,
@@ -290,25 +293,27 @@ class GitHub:
         # GitHub's advisories `cwes` filter expects bare numbers (89), not "CWE-89".
         if cwes and all(re.fullmatch(r"CWE-[1-9][0-9]*", cwe) for cwe in cwes):
             parameters["cwes"] = ",".join(sorted((cwe.removeprefix("CWE-") for cwe in cwes), key=int))
-        rows = self.get("/advisories?" + urlencode(parameters))
+        rows = self.get(next_url or "/advisories?" + urlencode(parameters))
         if not isinstance(rows, list):
             raise ValueError("invalid_advisory_page")
-        return {"rows": rows, "done": len(rows) < 100}
+        link = self.transport.response_headers.get("Link", "")
+        match = re.search(r'<([^>]+)>;\s*rel="next"', link)
+        return {"rows": rows, "next_url": match[1] if match else None}
 
     def advisories(self):
         """Enumerate the candidate population, deduplicating multi-ecosystem advisories."""
         seen = set()
         for ecosystem in ECOSYSTEMS:
-            page = 1
+            next_url = None
             while True:
-                result = self.advisory_page(page, ecosystem=ecosystem)
+                result = self.advisory_page(next_url, ecosystem=ecosystem)
                 for row in result["rows"]:
                     if row["ghsa_id"] not in seen:
                         seen.add(row["ghsa_id"])
                         yield row
-                if result["done"]:
+                next_url = result["next_url"]
+                if next_url is None:
                     break
-                page += 1
 
 
 class OSV:

tests/test_unseen_draft.py

@@ -139,12 +139,14 @@ class FakeAPI:
         # P is MIT but actual introducing head is GPL: public output must use V.
         return {"text": "", "spdx_id": "MIT" if ref == "b" * 40 else "GPL-3.0", "path": "LICENSE"}
 
-    def advisory_page(self, page, *, ecosystem):
+    def advisory_page(self, next_url=None, *, ecosystem):
         self.calls += 1
-        self.advisory_pages.append((ecosystem, page))
+        self.advisory_pages.append((ecosystem, next_url))
         population = list(self.advisories()) if ecosystem == "pip" else []
-        rows = population[(page - 1) * 100 : page * 100]
-        return {"rows": rows, "done": len(rows) < 100}
+        offset = int(next_url.rsplit("=", 1)[1]) if next_url else 0
+        rows = population[offset : offset + 100]
+        following = f"https://api.github.com/advisories?after={offset + 100}" if offset + 100 < len(population) else None
+        return {"rows": rows, "next_url": following}
 
     def advisories(self):
         for i in range(self.count):
@@ -460,8 +462,8 @@ def test_max_hours_stops_after_completed_page(tmp_path):
     now = [0.0]
 
     class TimedAPI(FakeAPI):
-        def advisory_page(self, page, *, ecosystem):
-            result = super().advisory_page(page, ecosystem=ecosystem)
+        def advisory_page(self, next_url=None, *, ecosystem):
+            result = super().advisory_page(next_url, ecosystem=ecosystem)
             now[0] += 3601
             return result
 
@@ -544,9 +546,9 @@ def test_advisory_journal_reduces_nested_blobs_and_resumes_by_page(tmp_path):
     class Pages:
         calls = []
 
-        def advisory_page(self, page, *, ecosystem):
-            self.calls.append((ecosystem, page))
-            return {"rows": [raw], "done": True, "unused": "raw page blob"}
+        def advisory_page(self, next_url=None, *, ecosystem):
+            self.calls.append((ecosystem, next_url))
+            return {"rows": [raw], "next_url": None, "unused": "raw page blob"}
 
     api, used = Pages(), e.UsedSet()
     parameters = d.inputs(used, 1, set(used.sources), FakeClones())
@@ -556,13 +558,13 @@ def test_advisory_journal_reduces_nested_blobs_and_resumes_by_page(tmp_path):
         assert rows == [expected]
     saved = [json.loads(line) for line in (path / "advisories.jsonl").read_bytes().splitlines()]
     assert [row["id"] for row in saved] == [f"{eco}:1" for eco in s.ECOSYSTEMS]
-    assert all(row["value"] == {"rows": [expected], "done": True} for row in saved)
+    assert all(row["value"] == {"rows": [expected], "next_url": None} for row in saved)
     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 == [(eco, 1) for eco in s.ECOSYSTEMS] and resumed == rows
+    assert api.calls == [(eco, None) for eco in s.ECOSYSTEMS] and resumed == rows
 
     class Bracket(FakeOSV):
         def get(self, identifier):
@@ -603,12 +605,12 @@ def test_reduced_advisories_preserve_all_fix_exclusions_and_selection():
             s.candidate(s.reduce_advisory(raw), FakeAPI(), FakeOSV())
 
 
-@pytest.mark.parametrize("old_version", [1, 2])
+@pytest.mark.parametrize("old_version", [1, 2, 3])
 def test_old_advisory_journal_version_refused_before_calls(tmp_path, old_version):
     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"] == 3
+    assert parameters["journal_version"] == 4
     with d.Journal(path).open({**parameters, "journal_version": old_version}) as journal:
         journal.put("advisories", "1", {"rows": list(api.advisories()), "done": True})
     report = d.run(
@@ -620,29 +622,30 @@ def test_old_advisory_journal_version_refused_before_calls(tmp_path, old_version
 
 def test_enumerate_ecosystem_pages_dedupes_and_resumes_partial_stream(tmp_path):
     rows = list(FakeAPI(3).advisories())
+    cursor = "https://api.github.com/advisories?ecosystem=pip&after=opaque-cursor"
     pages = {
-        ("pip", 1): {"rows": [rows[0]], "done": False},
-        ("pip", 2): {"rows": [rows[1]], "done": True},
-        ("npm", 1): {"rows": [rows[0], rows[2]], "done": True},
-        ("maven", 1): {"rows": [], "done": True},
-        ("composer", 1): {"rows": [], "done": True},
+        ("pip", None): {"rows": [rows[0]], "next_url": cursor},
+        ("pip", cursor): {"rows": [rows[0], rows[1]], "next_url": None},
+        ("npm", None): {"rows": [rows[0], rows[2]], "next_url": None},
+        ("maven", None): {"rows": [], "next_url": None},
+        ("composer", None): {"rows": [], "next_url": None},
     }
     calls = []
 
     class Pages:
-        def advisory_page(self, page, *, ecosystem):
-            calls.append((ecosystem, page))
-            return pages[ecosystem, page]
+        def advisory_page(self, next_url=None, *, ecosystem):
+            calls.append((ecosystem, next_url))
+            return pages[ecosystem, next_url]
 
     path = tmp_path / "journal"
-    with d.Journal(path).open({"journal_version": 3}) as journal:
-        journal.put("advisories", "pip:1", pages["pip", 1])
-    with d.Journal(path).open({"journal_version": 3}) as journal:
+    with d.Journal(path).open({"journal_version": 4}) as journal:
+        journal.put("advisories", "pip:1", pages["pip", None])
+    with d.Journal(path).open({"journal_version": 4}) as journal:
         result = d.enumerate_advisories(Pages(), journal)
         assert [row["ghsa_id"] for row in result] == [row["ghsa_id"] for row in rows]
-        assert list(journal.rows["advisories"]) == [f"{eco}:{page}" for eco, page in pages]
+        assert list(journal.rows["advisories"]) == ["pip:1", "pip:2", "npm:1", "maven:1", "composer:1"]
     assert calls == list(pages)[1:]
-    with d.Journal(path).open({"journal_version": 3}) as journal:
+    with d.Journal(path).open({"journal_version": 4}) as journal:
         assert d.enumerate_advisories(Pages(), journal) == result
     assert calls == list(pages)[1:]
 
@@ -660,11 +663,15 @@ def test_filtered_http_population_reaches_draft_floor_with_bounded_pages(tmp_pat
             assert query["ecosystem"][0] in s.ECOSYSTEMS
             assert query["published"] == [">=2019-01-01"]
             assert "89" in query["cwes"][0].split(",")
-            eco, page = query["ecosystem"][0], int(query["page"][0])
-            pages.append((eco, page))
-            assert len(pages) <= 7
+            eco, offset = query["ecosystem"][0], int(query.get("after", ["0"])[0])
+            assert "page" not in query
+            pages.append((eco, offset))
+            assert len(pages) <= 6
             rows = population if eco == "pip" else []
-            return rows[(page - 1) * 100 : page * 100]
+            self.response_headers = {}
+            if offset + 100 < len(rows):
+                self.response_headers = {"Link": f'<{url.split("&after=")[0]}&after={offset + 100}>; rel="next"'}
+            return rows[offset : offset + 100]
 
     api, used = FakeAPI(), e.UsedSet()
     http = s.GitHub("fake", FilteredTransport())
@@ -674,4 +681,4 @@ def test_filtered_http_population_reaches_draft_floor_with_bounded_pages(tmp_pat
     report = d.run(tmp_path, github=api, osv=FakeOSV(), clones=FakeClones(), used=used, seed=7, known_empty=set(used.sources))
     assert report["status"] == "draft" and report["repositories"] == 300
     assert report["advisories"] == 300
-    assert pages == [("pip", n) for n in range(1, 5)] + [(eco, 1) for eco in s.ECOSYSTEMS[1:]]
+    assert pages == [("pip", n) for n in (0, 100, 200)] + [(eco, 0) for eco in s.ECOSYSTEMS[1:]]

tests/test_unseen_source.py

@@ -114,6 +114,8 @@ def test_http_adapters_with_fake_transport():
     from urllib.parse import parse_qs, urlsplit
 
     class Transport:
+        response_headers = {}
+
         def __init__(self):
             self.calls = []
 
@@ -121,9 +123,10 @@ def test_http_adapters_with_fake_transport():
             self.calls.append((url, headers))
             parsed = urlsplit(url)
             if parsed.path == "/advisories":
-                page = int(parse_qs(parsed.query)["page"][0])
+                after = parse_qs(parsed.query).get("after")
+                self.response_headers = {} if after else {"Link": f'<{url}&after=cursor>; rel="next"'}
                 return [
-                    {**advisory(), "ghsa_id": f"GHSA-test-test-{i:04d}"} for i in range(0 if page == 1 else 100, 100 if page == 1 else 101)
+                    {**advisory(), "ghsa_id": f"GHSA-test-test-{i:04d}"} for i in range(0 if not after else 100, 100 if not after else 101)
                 ]
             if "/license" in parsed.path:
                 assert parse_qs(parsed.query)["ref"] == ["b" * 40]
@@ -150,6 +153,8 @@ def test_http_adapters_with_fake_transport():
 
 def test_http_failures_are_not_empty_results():
     class Transport:
+        response_headers = {}
+
         def get(self, url, headers):
             raise ValueError("http_failure")
 
@@ -283,12 +288,13 @@ def test_advisory_page_filters_match_candidate_population(ecosystem):
     expected = {c for f in taxonomy.families if f.id in s.FAMILIES for c in f.cwes}
 
     class Transport:
+        response_headers = {}
+
         def get(self, url, headers):
             query = parse_qs(urlsplit(url).query)
             assert query == {
                 "type": ["reviewed"],
                 "per_page": ["100"],
-                "page": ["2"],
                 "sort": ["published"],
                 "direction": ["asc"],
                 "ecosystem": [ecosystem],
@@ -297,7 +303,7 @@ def test_advisory_page_filters_match_candidate_population(ecosystem):
             }
             return []
 
-    assert s.GitHub("fake", Transport()).advisory_page(2, ecosystem=ecosystem) == {"rows": [], "done": True}
+    assert s.GitHub("fake", Transport()).advisory_page(ecosystem=ecosystem) == {"rows": [], "next_url": None}
     # Every enumerated CWE is selectable at the inclusive date boundary.
     for cwe in expected:
         row = advisory()
@@ -317,6 +323,8 @@ def test_advisory_page_omits_entire_cwe_filter_for_non_exact_taxonomy(monkeypatc
     monkeypatch.setattr(s, "load_families", lambda: taxonomy)
 
     class Transport:
+        response_headers = {}
+
         def get(self, url, headers):
             query = parse_qs(urlsplit(url).query)
             assert "cwes" not in query
@@ -324,4 +332,61 @@ def test_advisory_page_omits_entire_cwe_filter_for_non_exact_taxonomy(monkeypatc
             assert query["ecosystem"] == ["pip"]
             return []
 
-    s.GitHub("fake", Transport()).advisory_page(1, ecosystem="pip")
+    s.GitHub("fake", Transport()).advisory_page(ecosystem="pip")
+
+
+@pytest.mark.parametrize("journaled", [False, True])
+def test_cursor_walk_ignores_page_numbers_and_stops_without_next(tmp_path, journaled):
+    import io
+    from email.message import Message
+    from urllib.parse import parse_qs, urlencode, urlsplit
+
+    calls = []
+
+    def opener(request, timeout):
+        query = parse_qs(urlsplit(request.full_url).query)
+        eco = query["ecosystem"][0]
+        after = query.get("after", [None])[0]
+        calls.append((eco, after))
+        assert len(calls) <= 12  # Fail loudly if a page-number loop repeats forever.
+        # Like /advisories, ignore any page parameter. Only the opaque cursor advances.
+        offset = 100 if after == "opaque+/=" else 0
+        rows = [{**advisory(), "ghsa_id": f"GHSA-test-test-{i:04d}"} for i in range(offset, offset + 100)]
+        response = io.BytesIO(json.dumps(rows).encode())
+        response.headers = Message()
+        if after is None:
+            following = "https://api.github.com/advisories?" + urlencode({**query, "after": ["opaque+/="]}, doseq=True)
+            response.headers["link"] = f'<https://api.github.com/advisories?before=other>; rel="prev", <{following}>; rel="next"'
+        else:
+            # A full last page with a Link header, but no next relation, is exhausted.
+            response.headers["link"] = '<https://api.github.com/advisories?before=other>; rel="prev"'
+        return response
+
+    transport = s.JSONTransport(opener=opener)
+    first = "https://api.github.com/advisories?ecosystem=pip"
+    assert transport.get(first + "&page=1", {}) == transport.get(first + "&page=100", {})
+    calls.clear()
+    api = s.GitHub("fake", transport)
+    if journaled:
+        draft = importlib.import_module("benchmarks.unseen.draft")
+        with draft.Journal(tmp_path / "journal").open({"journal_version": 4}) as journal:
+            rows = draft.enumerate_advisories(api, journal)
+    else:
+        rows = list(api.advisories())
+    assert [row["ghsa_id"] for row in rows] == [f"GHSA-test-test-{i:04d}" for i in range(200)]
+    assert calls == [(eco, cursor) for eco in s.ECOSYSTEMS for cursor in (None, "opaque+/=")]
+
+
+def test_transport_replaces_last_successful_response_headers():
+    import io
+
+    responses = [io.BytesIO(b"{}"), io.BytesIO(b'{"broken":'), io.BytesIO(b"[]")]
+    responses[0].headers = {"Link": '<https://api.github.com/advisories?after=cursor>; rel="next"'}
+    responses[1].headers = {"Link": "failed response"}
+    waits = []
+    transport = s.JSONTransport(opener=lambda *a, **kw: responses.pop(0), sleeper=waits.append)
+    assert transport.get("https://example.invalid", {}) == {}
+    assert "cursor" in transport.response_headers["Link"]
+    assert transport.get("https://example.invalid", {}) == []
+    assert transport.response_headers == {}
+    assert waits == [1]