From 6241edfe713aac8c765ac394c9a13f56f84f6bdc Mon Sep 17 00:00:00 2001 From: woksin Date: Mon, 5 Oct 2026 23:22:49 +0200 Subject: [PATCH 1/2] Fix usage report coverage and paginated job prioritization --- .github/scripts/actions-usage-report.py | 54 ++++++++---- .../tests/actions-usage-report.test.py | 87 ++++++++++++++++++- 2 files changed, 122 insertions(+), 19 deletions(-) diff --git a/.github/scripts/actions-usage-report.py b/.github/scripts/actions-usage-report.py index cdbc1d2..b058f13 100644 --- a/.github/scripts/actions-usage-report.py +++ b/.github/scripts/actions-usage-report.py @@ -136,6 +136,7 @@ def inventory(repo): runs, candidates, seen_runs, failures = [], [], set(), Counter() inventory_errors = [] cursor = until + completed_window = False try: while cursor > since: window_start = max(cursor - dt.timedelta(days=1), since) @@ -148,32 +149,42 @@ def inventory(repo): # Public repositories use hosted runners and need run-level data only. if visibility == "private": candidates.append((full, visibility, run)) + completed_window = True except CollectionStopped: raise except APIError: 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 completed_window: + 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 {cursor.strftime('%Y-%m-%dT%H:%M:%SZ')} are 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 +192,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 +212,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..746ab15 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,88 @@ 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_no_completed_run_window_reports_all_runs_missing(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} for i in range(100)]} + data = report.collect(get, "Cratis", SINCE, UNTIL, workers=1) + 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_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 are 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 +588,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 are 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])) From cfc5d0b4c6d6f0ce78e34deb52be0e8f6241ae54 Mon Sep 17 00:00:00 2001 From: woksin Date: Mon, 5 Oct 2026 23:29:33 +0200 Subject: [PATCH 2/2] Fix partial run inventory coverage messages --- .github/scripts/actions-usage-report.py | 10 +++-- .../tests/actions-usage-report.test.py | 40 ++++++++++++++++--- 2 files changed, 41 insertions(+), 9 deletions(-) diff --git a/.github/scripts/actions-usage-report.py b/.github/scripts/actions-usage-report.py index b058f13..2355f98 100644 --- a/.github/scripts/actions-usage-report.py +++ b/.github/scripts/actions-usage-report.py @@ -136,12 +136,15 @@ def inventory(repo): runs, candidates, seen_runs, failures = [], [], set(), Counter() inventory_errors = [] cursor = until - completed_window = False 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"]) @@ -149,18 +152,17 @@ def inventory(repo): # Public repositories use hosted runners and need run-level data only. if visibility == "private": candidates.append((full, visibility, run)) - completed_window = True except CollectionStopped: raise except APIError: failures["Run inventory incomplete"] += 1 cursor = window_start except CollectionStopped: - if not completed_window: + 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 {cursor.strftime('%Y-%m-%dT%H:%M:%SZ')} are missing.") + 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 diff --git a/.github/scripts/tests/actions-usage-report.test.py b/.github/scripts/tests/actions-usage-report.test.py index 746ab15..aa16179 100644 --- a/.github/scripts/tests/actions-usage-report.test.py +++ b/.github/scripts/tests/actions-usage-report.test.py @@ -544,7 +544,7 @@ def completed_newest_last(pending, return_when): 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_no_completed_run_window_reports_all_runs_missing(self): + 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 = [] @@ -556,13 +556,43 @@ def get(endpoint): return [PRIVATE] if endpoint == "orgs/Cratis": return {"total_private_repos": 1} - return {"workflow_runs": [{"id": i} for i in range(100)]} + 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) - self.assertIn("Run inventory not started for Cratis/Private; all runs are missing.", data[4]) + 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": @@ -574,7 +604,7 @@ def get(endpoint): 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 are missing.", data[4]) + "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): @@ -589,7 +619,7 @@ def get(endpoint): return {"workflow_runs": [{"id": day}]} data = report.collect(get, "Cratis", SINCE, UNTIL, workers=1) self.assertIn("Run inventory stopped before completing Cratis/Private; " - "runs created before 2026-09-29T00:00:00Z are missing.", data[4]) + "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]))