Fix usage report missing-data coverage and continuation priority

.github/scripts/actions-usage-report.py

@@ -138,9 +138,13 @@ def inventory(repo):
         cursor = until
         try:
             while cursor > since:
+                missing_before = cursor
                 window_start = max(cursor - dt.timedelta(days=1), since)
                 try:
                     for run in window_runs(request, full, window_start, cursor):
+                        created = timestamp(run.get("created_at"))
+                        if created is not None:
+                            missing_before = min(missing_before, created)
                         if run["id"] in seen_runs:
                             continue  # GitHub's range endpoints are inclusive.
                         seen_runs.add(run["id"])
@@ -154,33 +158,47 @@ def inventory(repo):
                     failures["Run inventory incomplete"] += 1
                 cursor = window_start
         except CollectionStopped:
-            inventory_errors.append(f"Run inventory stopped before completing {full}; older runs may be missing.")
+            if not runs and cursor == until:
+                inventory_errors.append(f"Run inventory not started for {full}; all runs are missing.")
+            else:
+                inventory_errors.append(f"Run inventory stopped before completing {full}; "
+                                        f"runs created before {missing_before.strftime('%Y-%m-%dT%H:%M:%SZ')} may be missing.")
         inventory_errors.extend(f"{kind} for {full}: {count} daily windows." for kind, count in failures.items())
         return runs, candidates, inventory_errors
 
-    def job_inventory(candidate):
+    def job_inventory(candidate, page):
         full, visibility, run = candidate
         found = []
         try:
-            # Each run is independent: a busy repository uses the entire bounded
-            # pool instead of serializing thousands of job requests on one worker.
-            # The scheduler reserves page one in priority order before submission;
-            # workers reserve any additional pages from the same shared budget.
-            get_jobs = lambda endpoint: request.fetch(endpoint) if endpoint.endswith("&page=1") else request(endpoint)
-            for job in pages(get_jobs, f"repos/{full}/actions/runs/{run['id']}/jobs?filter=all", "jobs"):
+            # Fetch only the page reserved by the scheduler; continuation pages
+            # compete with other runs for the same newest-first request budget.
+            response = request.fetch(f"repos/{full}/actions/runs/{run['id']}/jobs?filter=all&per_page=100&page={page}")
+            items = response["jobs"]
+            if not isinstance(items, list):
+                raise APIError("Invalid API listing")
+            for job in items:
                 found.append((full, visibility, run.get("name") or "Unnamed workflow", job))
+            if len(items) >= 100:
+                if page == 1000:
+                    raise APIError("API pagination limit exceeded")
+                return found, None, page + 1
         except CollectionStopped:
-            return found, "Job inventory stopped before completing listed private runs"
+            return found, "Job inventory stopped before completing listed private runs", None
         except (APIError, KeyError, ValueError, UnicodeError):
-            return found, "Job or runner inventory incomplete"
-        return found, None
+            return found, "Job or runner inventory incomplete", None
+        return found, None, None
 
     active = [repo for repo in repos if not repo.get("archived")]
     active.sort(key=lambda repo: not (repo.get("private") or repo.get("visibility") == "private"))
     runs, jobs, seen_jobs, job_failures = [], [], set(), Counter()
     private_repos = deque(repo for repo in active if repo.get("private") or repo.get("visibility") == "private")
     public_repos = deque(repo for repo in active if not (repo.get("private") or repo.get("visibility") == "private"))
     candidates, pending, private_inventories = [], {}, set()
+
+    def queue_jobs(candidate, page):
+        created = timestamp(candidate[2].get("created_at")) or since
+        heapq.heappush(candidates, (-created.timestamp(), candidate[0], candidate[2]["id"], page, candidate))
+
     with ThreadPoolExecutor(max_workers=workers) as pool:
         while private_repos or public_repos or candidates or pending:
             # Inventory every private repository before jobs so a budget cut drops
@@ -196,28 +214,30 @@ def job_inventory(candidate):
                 elif public_repos:
                     pending[pool.submit(inventory, public_repos.popleft())] = None
                 elif candidates:
-                    candidate = heapq.heappop(candidates)[3]
+                    page, candidate = heapq.heappop(candidates)[3:]
                     try:
                         request.reserve()
                     except CollectionStopped:
                         job_failures[(candidate[0], "Job inventory stopped before completing listed private runs")] += 1
                         continue
-                    pending[pool.submit(job_inventory, candidate)] = candidate
+                    pending[pool.submit(job_inventory, candidate, page)] = (candidate, page)
                 else:
                     break
             completed, _ = wait(pending, return_when=FIRST_COMPLETED)
             for future in completed:
-                candidate = pending.pop(future)
-                if candidate is None:
+                task = pending.pop(future)
+                if task is None:
                     private_inventories.discard(future)
                     repo_runs, repo_candidates, repo_errors = future.result()
                     runs.extend(repo_runs)
                     for entry in repo_candidates:
-                        created = timestamp(entry[2].get("created_at")) or since
-                        heapq.heappush(candidates, (-created.timestamp(), entry[0], entry[2]["id"], entry))
+                        queue_jobs(entry, 1)
                     errors.extend(repo_errors)
                     continue
-                found, failure = future.result()
+                candidate, _ = task
+                found, failure, continuation = future.result()
+                if continuation is not None:
+                    queue_jobs(candidate, continuation)
                 if failure:
                     job_failures[(candidate[0], failure)] += 1
                 for entry in found:

.github/scripts/tests/actions-usage-report.test.py

@@ -1,12 +1,14 @@
 # Copyright (c) Cratis. All rights reserved.
 # Licensed under the MIT license. See LICENSE file in the project root for full license information.
 """Offline usage-report fixtures; no token or live GitHub calls."""
+from concurrent.futures import Future
 import datetime as dt
 import importlib.util
 from pathlib import Path
 import threading
 import time
 import unittest
+from unittest.mock import patch
 
 spec = importlib.util.spec_from_file_location("report", Path(__file__).parents[1] / "actions-usage-report.py")
 report = importlib.util.module_from_spec(spec)
@@ -493,6 +495,118 @@ def get(endpoint):
                     self.assertIn("Job inventory stopped before completing listed private runs "
                                   f"for {repo['full_name']}: 5 runs.", data[4])
 
+    def test_job_continuations_keep_run_priority_with_immediate_page_arrivals(self):
+        # Finish submitted pages immediately so both first-page results reach
+        # the scheduler together, without relying on worker timing.
+        class ImmediateExecutor:
+            def __init__(self, max_workers):
+                pass
+
+            def __enter__(self):
+                return self
+
+            def __exit__(self, *args):
+                pass
+
+            def submit(self, function, *args):
+                future = Future()
+                future.set_result(function(*args))
+                return future
+
+        calls = []
+        def get(endpoint):
+            if endpoint == "rate_limit":
+                # Metadata, one daily window, two first pages, one continuation.
+                return {"resources": {"core": {"remaining": 106}}}
+            if endpoint.startswith("orgs/Cratis/repos?"):
+                return [PRIVATE]
+            if endpoint == "orgs/Cratis":
+                return {"total_private_repos": 1}
+            if "/jobs?" in endpoint:
+                identity = int(endpoint.split("/runs/")[1].split("/")[0])
+                page = int(endpoint.rsplit("=", 1)[1])
+                calls.append((identity, page))
+                if page == 1:
+                    return {"jobs": [job(identity * 1000 + i) for i in range(100)]}
+                return {"jobs": [job(identity * 1000 + 100)]}
+            return {"workflow_runs": [{"id": day, "created_at": f"2026-09-{day}T12:00:00Z"}
+                                      for day in (29, 30, 28)]}
+
+        def completed_newest_last(pending, return_when):
+            # Hand back the older continuation first; arrival order must not
+            # decide which continuation gets the final budget reservation.
+            return list(reversed(pending)), set()
+
+        with patch.object(report, "ThreadPoolExecutor", ImmediateExecutor), \
+                patch.object(report, "wait", completed_newest_last):
+            data = report.collect(get, "Cratis", UNTIL - dt.timedelta(days=1), UNTIL, workers=2)
+        self.assertEqual(calls, [(30, 1), (29, 1), (30, 2)])
+        self.assertEqual(len(data[3]), 201)
+        self.assertIn("Job inventory stopped before completing listed private runs for Cratis/Private: 2 runs.", data[4])
+
+    def test_first_window_budget_cut_distinguishes_unstarted_and_partial_inventories(self):
+        for partial_page in (False, True):
+            with self.subTest(partial_page=partial_page):
+                calls = []
+                def get(endpoint):
+                    calls.append(endpoint)
+                    if endpoint == "rate_limit":
+                        return {"resources": {"core": {"remaining": 103 if partial_page else 102}}}
+                    if endpoint.startswith("orgs/Cratis/repos?"):
+                        return [PRIVATE]
+                    if endpoint == "orgs/Cratis":
+                        return {"total_private_repos": 1}
+                    return {"workflow_runs": [{"id": i, "created_at": "2026-09-30T00:00:00Z"}
+                                             for i in range(100)]}
+                data = report.collect(get, "Cratis", SINCE, UNTIL, workers=1)
+                if partial_page:
+                    self.assertIn("Run inventory stopped before completing Cratis/Private; "
+                                  "runs created before 2026-09-30T00:00:00Z may be missing.", data[4])
+                    self.assertFalse(any("all runs are missing" in error for error in data[4]))
+                else:
+                    self.assertIn("Run inventory not started for Cratis/Private; all runs are missing.", data[4])
+                self.assertNotIn("older runs may be missing", "\n".join(data[4]))
+                self.assertEqual(len(data[2]), 100 if partial_page else 0)
+                self.assertEqual(len(calls), 4 if partial_page else 3)
+
+    def test_failed_run_windows_then_budget_cut_reports_partial_inventory_consistently(self):
+        for partial_page in (False, True):
+            with self.subTest(partial_page=partial_page):
+                def get(endpoint):
+                    if endpoint == "rate_limit":
+                        # Metadata, then two failing daily windows; no completed windows.
+                        return {"resources": {"core": {"remaining": 106 if partial_page else 104}}}
+                    if endpoint.startswith("orgs/Cratis/repos?"):
+                        return [PUBLIC]
+                    if endpoint == "orgs/Cratis":
+                        return {"total_private_repos": 0}
+                    if partial_page and endpoint.endswith("page=1"):
+                        day = int(endpoint.split("created=2026-09-")[1][:2])
+                        return {"workflow_runs": [{"id": day * 100 + i,
+                                                  "created_at": f"2026-09-{day}T12:00:00Z"}
+                                                 for i in range(100)]}
+                    raise report.APIError("denied")
+                data = report.collect(get, "Cratis", SINCE, UNTIL, workers=1)
+                self.assertEqual(len(data[2]), 200 if partial_page else 0)
+                self.assertIn("Run inventory stopped before completing Cratis/Public; "
+                              "runs created before 2026-09-29T00:00:00Z may be missing.", data[4])
+                self.assertIn("Run inventory incomplete for Cratis/Public: 2 daily windows.", data[4])
+                self.assertFalse(any("all runs are missing" in error for error in data[4]))
+
+    def test_empty_completed_run_window_reports_the_precise_missing_cursor(self):
+        def get(endpoint):
+            if endpoint == "rate_limit":
+                return {"resources": {"core": {"remaining": 103}}}
+            if endpoint.startswith("orgs/Cratis/repos?"):
+                return [PRIVATE]
+            if endpoint == "orgs/Cratis":
+                return {"total_private_repos": 1}
+            return {"workflow_runs": []}
+        data = report.collect(get, "Cratis", SINCE, UNTIL, workers=1)
+        self.assertIn("Run inventory stopped before completing Cratis/Private; "
+                      "runs created before 2026-09-30T00:00:00Z may be missing.", data[4])
+        self.assertFalse(any("all runs are missing" in error for error in data[4]))
+
     def test_run_inventory_stop_is_not_misreported_as_one_missing_run(self):
         def get(endpoint):
             if endpoint == "rate_limit":
@@ -504,7 +618,8 @@ def get(endpoint):
             day = int(endpoint.split("created=2026-09-")[1][:2])
             return {"workflow_runs": [{"id": day}]}
         data = report.collect(get, "Cratis", SINCE, UNTIL, workers=1)
-        self.assertIn("Run inventory stopped before completing Cratis/Private; older runs may be missing.", data[4])
+        self.assertIn("Run inventory stopped before completing Cratis/Private; "
+                      "runs created before 2026-09-29T00:00:00Z may be missing.", data[4])
         self.assertTrue(any("Job inventory stopped" in error and "2 runs" in error for error in data[4]))
         self.assertFalse(any("1 runs" in error for error in data[4]))