diff --git a/.github/scripts/actions-usage-report.py b/.github/scripts/actions-usage-report.py index cdbc1d2..2355f98 100644 --- a/.github/scripts/actions-usage-report.py +++ b/.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,26 +158,35 @@ 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")) @@ -181,6 +194,11 @@ def job_inventory(candidate): 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: diff --git a/.github/scripts/tests/actions-usage-report.test.py b/.github/scripts/tests/actions-usage-report.test.py index b0353b5..aa16179 100644 --- a/.github/scripts/tests/actions-usage-report.test.py +++ b/.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]))