Merge unseen-cursor-pagination: follow Link rel=next cursors

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]