From 23e2dba16e94a4965b6f885cac907e3d632bd740 Mon Sep 17 00:00:00 2001 From: Trask Stalnaker Date: Tue, 13 Oct 2026 13:08:34 -0700 Subject: [PATCH] Unblock GitHub Actions queue retries Use smaller jobs pages, share bounded pending retries between failed and unfinished runs, and recognize terminal job evidence when GitHub retains a stale status. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: dce070cb-ab76-4725-80eb-0db99ed545b4 --- .../scripts/github-actions-queue/collect.py | 72 ++- .../github-actions-queue/test_collect.py | 426 +++++++++++++++++- github-actions-queue/README.md | 28 +- 3 files changed, 504 insertions(+), 22 deletions(-) diff --git a/.github/scripts/github-actions-queue/collect.py b/.github/scripts/github-actions-queue/collect.py index d1786f7d417e..712c909b3fcc 100644 --- a/.github/scripts/github-actions-queue/collect.py +++ b/.github/scripts/github-actions-queue/collect.py @@ -23,10 +23,21 @@ API_ROOT = "https://api.github.com" API_VERSION = "2022-11-28" PAGE_SIZE = 100 +JOBS_PAGE_SIZE = 50 PENDING_RETRY_BATCH_SIZE = 25 +PENDING_FAILURE_RETRY_BATCH_SIZE = 5 STATE_VERSION = 1 USER_AGENT = "open-telemetry-github-actions-queue-collector" COLLECTION_ID_PATTERN = re.compile(r"^[A-Za-z0-9._-]+$") +TERMINAL_JOB_CONCLUSIONS = { + "success", + "failure", + "neutral", + "cancelled", + "skipped", + "timed_out", + "action_required", +} class ApiError(RuntimeError): @@ -461,7 +472,7 @@ def list_jobs( if not isinstance(total, int) or not isinstance(jobs, list): raise ApiError("workflow jobs response has an invalid shape") - pages = math.ceil(total / PAGE_SIZE) + pages = math.ceil(total / JOBS_PAGE_SIZE) for page in range(2, pages + 1): response = self._jobs_page(org, repository, run_id, page) page_jobs = response.get("jobs") @@ -488,7 +499,7 @@ def _jobs_page( f"/repos/{quote(org)}/{quote(repository)}/actions/runs/{run_id}/jobs", { "filter": "all", - "per_page": str(PAGE_SIZE), + "per_page": str(JOBS_PAGE_SIZE), "page": str(page), }, ) @@ -644,14 +655,7 @@ def _collect( for item in state.pending_runs if item["repository"] in current_repositories ] - pending_to_retry = sorted( - state.pending_runs, - key=lambda item: ( - item.get("last_attempt_at") or item.get("created_at") or "", - item["repository"], - item["run_id"], - ), - )[:PENDING_RETRY_BATCH_SIZE] + pending_to_retry = _pending_runs_to_retry(state.pending_runs) self._collect_pending_runs(state, pending_to_retry, records) while ( @@ -789,7 +793,9 @@ def _collect_pending_runs( else: _merge_pending_runs( state, - [_pending_run_attempt(item, self._now())], + [ + _pending_run_attempt(item, self._now(), error_resolved=True) + ], ) except RateLimitExhausted: raise @@ -835,7 +841,7 @@ def _collect_completed_run( run: dict[str, Any], ) -> tuple[list[dict[str, Any]], bool]: jobs = self._client.list_jobs(self._org, repository, run["id"]) - if any(job.get("status") != "completed" for job in jobs): + if any(not _job_is_terminal(job) for job in jobs): return [], False jobs = [job for job in jobs if job.get("runner_name")] collected_at = _format_instant(self._now()) @@ -892,6 +898,41 @@ def collect_live_and_backfill( ) +def _job_is_terminal(job: dict[str, Any]) -> bool: + return job.get("status") == "completed" or ( + job.get("status") == "in_progress" + and bool(job.get("completed_at")) + and job.get("conclusion") in TERMINAL_JOB_CONCLUSIONS + ) + + +def _pending_runs_to_retry( + pending: list[dict[str, Any]], +) -> list[dict[str, Any]]: + def priority(item: dict[str, Any]) -> tuple[str, str, int]: + return ( + item.get("last_attempt_at") or item.get("created_at") or "", + item["repository"], + item["run_id"], + ) + + ordered = sorted(pending, key=priority) + failed = [item for item in ordered if item.get("last_error")] + ordinary = [item for item in ordered if not item.get("last_error")] + failed_slots = max( + PENDING_FAILURE_RETRY_BATCH_SIZE, + PENDING_RETRY_BATCH_SIZE - len(ordinary), + ) + selected_failures = failed[:failed_slots] + return sorted( + [ + *selected_failures, + *ordinary[: PENDING_RETRY_BATCH_SIZE - len(selected_failures)], + ], + key=priority, + ) + + def _pending_run( repository: str, run: dict[str, Any], @@ -919,11 +960,16 @@ def _pending_run_failure( def _pending_run_attempt( item: dict[str, Any], attempted_at: datetime, + *, + error_resolved: bool = False, ) -> dict[str, Any]: - return { + attempted = { **item, "last_attempt_at": _format_instant(attempted_at), } + if error_resolved: + attempted.pop("last_error", None) + return attempted def _merge_pending_runs( diff --git a/.github/scripts/github-actions-queue/test_collect.py b/.github/scripts/github-actions-queue/test_collect.py index 6f00d7d44e14..8d799f5188f3 100644 --- a/.github/scripts/github-actions-queue/test_collect.py +++ b/.github/scripts/github-actions-queue/test_collect.py @@ -22,6 +22,7 @@ RateLimitExhausted, RateSnapshot, _job_record, + _pending_runs_to_retry, collect_live_and_backfill, load_backfill_state, load_state, @@ -104,12 +105,15 @@ def test_paginates_all_jobs_and_requests_all_attempts(self): def handler(url): query = parse_qs(urlparse(url).query) page = int(query["page"][0]) - start = (page - 1) * 100 - count = 100 if page < 3 else 50 + self.assertEqual(["50"], query["per_page"]) + start = (page - 1) * 50 return response( { "total_count": 250, - "jobs": [{"id": index} for index in range(start, start + count)], + "jobs": [ + {"id": index, "run_attempt": 1 if index < 100 else 2} + for index in range(start, start + 50) + ], } ) @@ -119,13 +123,81 @@ def handler(url): jobs = client.list_jobs("open-telemetry", "example", 7) self.assertEqual(250, len(jobs)) + self.assertEqual({1, 2}, {job["run_attempt"] for job in jobs}) job_urls = [url for url in transport.urls if "/jobs?" in url] - self.assertEqual(3, len(job_urls)) + self.assertEqual(5, len(job_urls)) self.assertTrue(all("filter=all" in url for url in job_urls)) + def test_job_pagination_keeps_partial_last_page(self): + def handler(url): + query = parse_qs(urlparse(url).query) + page = int(query["page"][0]) + start = (page - 1) * 50 + return response( + { + "total_count": 164, + "jobs": [ + {"id": index} + for index in range(start, min(start + 50, 164)) + ], + } + ) + + transport = FakeTransport(handler) + client = GitHubClient("token", transport=transport) + + jobs = client.list_jobs("open-telemetry", "example", 7) + + self.assertEqual(list(range(164)), [job["id"] for job in jobs]) + self.assertEqual( + ["1", "2", "3", "4"], + [ + parse_qs(urlparse(url).query)["page"][0] + for url in transport.urls + if "/jobs?" in url + ], + ) + + def test_job_pagination_rejects_missing_or_duplicate_jobs(self): + for second_page in ([{"id": 50}], [{"id": 0}, {"id": 50}]): + with self.subTest(second_page=second_page): + + def handler(url): + query = parse_qs(urlparse(url).query) + return response( + { + "total_count": 52, + "jobs": ( + [{"id": index} for index in range(50)] + if query["page"] == ["1"] + else second_page + ), + } + ) + + client = GitHubClient("token", transport=FakeTransport(handler)) + + with self.assertRaisesRegex(ApiError, "returned 51 of 52 jobs"): + client.list_jobs("open-telemetry", "example", 7) + + def test_job_pagination_respects_request_budget_without_returning_partial_jobs(self): + transport = FakeTransport( + lambda _url: response( + {"total_count": 100, "jobs": [{"id": index} for index in range(50)]} + ) + ) + client = GitHubClient("token", transport=transport, max_requests=2) + + with self.assertRaisesRegex(CollectionBudgetExhausted, "request_limit"): + client.list_jobs("open-telemetry", "example", 7) + + self.assertEqual(2, client.request_count) + self.assertEqual(2, len(transport.urls)) + def test_splits_run_searches_above_one_thousand(self): def handler(url): query = parse_qs(urlparse(url).query) + self.assertEqual(["100"], query["per_page"]) created = query["created"][0] if created.startswith("2026-09-15T00:00:00Z") and created.endswith( "2026-09-15T00:59:59Z" @@ -429,6 +501,99 @@ def test_unknown_head_repository_preserves_unknown_fork_origin(self): self.assertIsNone(record["from_fork"]) +class PendingRunSelectionTest(unittest.TestCase): + def test_shares_retry_budget_without_starving_either_queue(self): + for ordinary_count, failed_count, ordinary_slots, failed_slots in ( + (30, 30, 20, 5), + (30, 2, 23, 2), + (2, 30, 2, 23), + (0, 30, 0, 25), + (30, 0, 25, 0), + (2, 2, 2, 2), + (0, 0, 0, 0), + ): + with self.subTest(ordinary=ordinary_count, failed=failed_count): + pending = [ + *[ + { + "repository": "a", + "run_id": run_id, + "created_at": "2026-01-01T00:00:00Z", + } + for run_id in range(1, ordinary_count + 1) + ], + *[ + { + "repository": "a", + "run_id": run_id, + "created_at": "2026-01-01T00:00:00Z", + "last_attempt_at": "2026-09-15T00:00:00Z", + "last_error": "GitHub API returned 502", + "failures": 3, + } + for run_id in range(101, failed_count + 101) + ], + ] + + selected = _pending_runs_to_retry(list(reversed(pending))) + + self.assertEqual( + [*range(1, ordinary_slots + 1), *range(101, failed_slots + 101)], + [item["run_id"] for item in selected], + ) + + def test_ordinary_retries_keep_their_share_when_failures_are_older(self): + pending = [ + *[ + { + "repository": "a", + "run_id": run_id, + "created_at": "2026-09-15T00:00:00Z", + } + for run_id in range(1, 31) + ], + *[ + { + "repository": "a", + "run_id": run_id, + "created_at": "2026-01-01T00:00:00Z", + "last_error": "GitHub API returned 502", + } + for run_id in range(101, 131) + ], + ] + + self.assertEqual( + [*range(101, 106), *range(1, 21)], + [item["run_id"] for item in _pending_runs_to_retry(pending)], + ) + + def test_uses_last_error_instead_of_historical_failure_count(self): + pending = [ + { + "repository": "a", + "run_id": run_id, + "created_at": "2026-01-01T00:00:00Z", + "failures": 3, + } + for run_id in range(1, 31) + ] + pending.append( + { + "repository": "a", + "run_id": 101, + "created_at": "2026-09-15T00:00:00Z", + "last_error": "GitHub API returned 502", + "failures": 3, + } + ) + + self.assertEqual( + [*range(1, 25), 101], + [item["run_id"] for item in _pending_runs_to_retry(pending)], + ) + + class FakeClient: def __init__(self): self.rate = RateSnapshot(5000, 4990, 2000000000) @@ -730,6 +895,142 @@ def get_workflow_run(self, org, repository, run_id): self.assertEqual([10], [record["job_id"] for record in result.records]) self.assertEqual([10], [item["run_id"] for item in state.pending_runs]) + def test_failed_retries_rotate_alongside_old_historical_pending_runs(self): + class TrackingClient(FakeClient): + def __init__(self): + super().__init__() + self.retried = [] + + def get_workflow_run(self, _org, _repository, run_id): + self.retried.append(run_id) + return {"id": run_id, "status": "in_progress"} + + client = TrackingClient() + collector = QueueCollector( + client, + org="open-telemetry", + now=lambda: datetime(2026, 9, 15, 1, tzinfo=UTC), + ) + state = BackfillState( + initial_cursor="2026-09-15T01:00:00Z", + target="2026-01-01T00:00:00Z", + cursor="2026-01-01T00:00:00Z", + pending_runs=[ + *[ + { + "repository": "a", + "run_id": run_id, + "created_at": "2026-01-01T00:00:00Z", + } + for run_id in range(1, 41) + ], + *[ + { + "repository": "a", + "run_id": run_id, + "created_at": "2026-01-01T00:00:00Z", + "last_attempt_at": "2026-09-15T00:00:00Z", + "last_error": "GitHub API returned 502", + } + for run_id in range(101, 111) + ], + ], + ) + + first = collector.collect_backfill(state, selected_repositories=["a"]) + second = collector.collect_backfill(state, selected_repositories=["a"]) + + self.assertFalse(first.paused) + self.assertFalse(second.paused) + self.assertEqual( + [*range(1, 21), *range(101, 106), *range(21, 41), *range(106, 111)], + client.retried, + ) + + def test_successful_pending_job_lookup_clears_only_the_current_error(self): + class UnfinishedJobClient(FakeClient): + def list_jobs(self, org, repository, run_id): + return [ + { + **super().list_jobs(org, repository, run_id)[0], + "status": "in_progress", + "conclusion": None, + "completed_at": None, + } + ] + + client = UnfinishedJobClient() + client.completed_runs[("a", 9)] = {"id": 9, "status": "completed"} + collector = QueueCollector( + client, + org="open-telemetry", + now=lambda: datetime(2026, 9, 15, 1, tzinfo=UTC), + ) + state = CollectorState( + cursor="2026-09-15T01:00:00Z", + pending_runs=[ + { + "repository": "a", + "run_id": 9, + "created_at": "2026-09-14T23:00:00Z", + "last_error": "GitHub API returned 502", + "failures": 3, + } + ], + ) + + result = collector.collect(state, selected_repositories=["a"]) + + self.assertEqual([], result.records) + self.assertEqual( + [ + { + "repository": "a", + "run_id": 9, + "created_at": "2026-09-14T23:00:00Z", + "last_attempt_at": "2026-09-15T01:00:00Z", + "failures": 3, + } + ], + state.pending_runs, + ) + + client.fail_jobs_for_run = 9 + collector.collect(state, selected_repositories=["a"]) + + self.assertEqual(4, state.pending_runs[0]["failures"]) + self.assertEqual("GitHub API returned 502", state.pending_runs[0]["last_error"]) + + def test_unfinished_run_keeps_the_unverified_job_error(self): + client = FakeClient() + client.completed_runs[("a", 9)] = {"id": 9, "status": "in_progress"} + collector = QueueCollector( + client, + org="open-telemetry", + now=lambda: datetime(2026, 9, 15, 1, tzinfo=UTC), + ) + state = CollectorState( + cursor="2026-09-15T01:00:00Z", + pending_runs=[ + { + "repository": "a", + "run_id": 9, + "created_at": "2026-09-14T23:00:00Z", + "last_error": "GitHub API returned 502", + "failures": 3, + } + ], + ) + + result = collector.collect(state, selected_repositories=["a"]) + + self.assertEqual([], result.records) + self.assertEqual("GitHub API returned 502", state.pending_runs[0]["last_error"]) + self.assertEqual(3, state.pending_runs[0]["failures"]) + self.assertEqual( + "2026-09-15T01:00:00Z", state.pending_runs[0]["last_attempt_at"] + ) + def test_job_api_failure_does_not_block_window_progress(self): client = FakeClient() client.fail_jobs_for_run = 100 @@ -752,6 +1053,82 @@ def test_job_api_failure_does_not_block_window_progress(self): self.assertEqual(1, state.pending_runs[0]["failures"]) self.assertIn("502", state.pending_runs[0]["last_error"]) + def test_requires_terminal_evidence_for_stale_job_status(self): + for status, conclusion, completed_at, terminal in ( + ("in_progress", "success", "2026-09-15T00:12:00Z", True), + ("in_progress", "failure", "2026-09-15T00:12:00Z", True), + ("in_progress", "cancelled", "2026-09-15T00:12:00Z", True), + ("in_progress", "neutral", "2026-09-15T00:12:00Z", True), + ("in_progress", "skipped", "2026-09-15T00:12:00Z", True), + ("in_progress", "timed_out", "2026-09-15T00:12:00Z", True), + ("in_progress", "action_required", "2026-09-15T00:12:00Z", True), + ("in_progress", "success", None, False), + ("in_progress", None, "2026-09-15T00:12:00Z", False), + ("in_progress", "unknown", "2026-09-15T00:12:00Z", False), + ("queued", "success", "2026-09-15T00:12:00Z", False), + ("waiting", None, None, False), + ): + with self.subTest( + status=status, conclusion=conclusion, completed_at=completed_at + ): + + class StaleStatusClient(FakeClient): + def list_jobs(self, org, repository, run_id): + return [ + { + **super().list_jobs(org, repository, run_id)[0], + "status": status, + "conclusion": conclusion, + "completed_at": completed_at, + } + ] + + collector = QueueCollector( + StaleStatusClient(), + org="open-telemetry", + now=lambda: datetime(2026, 9, 15, 1, tzinfo=UTC), + ) + state = CollectorState(cursor="2026-09-15T00:00:00Z") + + result = collector.collect(state, selected_repositories=["a"]) + + self.assertEqual( + [101] if terminal else [], + [record["job_id"] for record in result.records], + ) + self.assertEqual(0 if terminal else 1, len(state.pending_runs)) + if terminal: + self.assertEqual(status, result.records[0]["job_status"]) + self.assertEqual(conclusion, result.records[0]["job_conclusion"]) + + def test_waits_for_every_attempt_even_when_other_jobs_have_stale_status(self): + class UnfinishedAttemptClient(FakeClient): + def list_jobs(self, org, repository, run_id): + job = super().list_jobs(org, repository, run_id)[0] + return [ + {**job, "run_attempt": 1, "status": "in_progress"}, + { + **job, + "id": job["id"] + 1, + "run_attempt": 2, + "status": "in_progress", + "conclusion": None, + "completed_at": None, + }, + ] + + collector = QueueCollector( + UnfinishedAttemptClient(), + org="open-telemetry", + now=lambda: datetime(2026, 9, 15, 1, tzinfo=UTC), + ) + state = CollectorState(cursor="2026-09-15T00:00:00Z") + + result = collector.collect(state, selected_repositories=["a"]) + + self.assertEqual([], result.records) + self.assertEqual([100], [item["run_id"] for item in state.pending_runs]) + def test_discards_negative_queue_jobs_without_matching(self): class RerunClient(FakeClient): def list_workflow_runs(self, _org, _repository, _start, _end): @@ -791,6 +1168,7 @@ def list_jobs(self, _org, _repository, _run_id): { **common, "id": 2, + "status": "in_progress", "run_attempt": 2, "created_at": "2026-09-15T00:20:00Z", "started_at": "2026-09-15T00:11:00Z", @@ -807,6 +1185,7 @@ def list_jobs(self, _org, _repository, _run_id): { **common, "id": 4, + "status": "in_progress", "name": "skipped", "run_attempt": 2, "created_at": "2026-09-15T00:20:00Z", @@ -818,6 +1197,7 @@ def list_jobs(self, _org, _repository, _run_id): { **common, "id": 5, + "status": "in_progress", "name": "unmatched", "run_attempt": 2, "created_at": "2026-09-15T00:20:00Z", @@ -1067,6 +1447,44 @@ def list_workflow_runs(self, org, repository, start, end): client.windows, ) + def test_live_pause_does_not_retry_historical_failures(self): + client = FakeClient() + client.pause_on_repo = "a" + collector = QueueCollector( + client, + org="open-telemetry", + now=lambda: datetime(2026, 9, 15, 2, tzinfo=UTC), + ) + live_state = CollectorState(cursor="2026-09-15T01:00:00Z") + backfill_state = BackfillState( + initial_cursor="2026-09-15T01:00:00Z", + target="2026-09-14T23:00:00Z", + cursor="2026-09-15T01:00:00Z", + pending_runs=[ + { + "repository": "a", + "run_id": 9, + "created_at": "2026-09-14T23:00:00Z", + "last_error": "GitHub API returned 502", + "failures": 3, + } + ], + ) + + result = collect_live_and_backfill( + collector, + live_state, + backfill_state, + now=datetime(2026, 9, 15, 2, tzinfo=UTC), + selected_repositories=["a"], + max_windows=2, + ) + + self.assertTrue(result.live.paused) + self.assertIsNone(result.backfill) + self.assertEqual(3, backfill_state.pending_runs[0]["failures"]) + self.assertNotIn("last_attempt_at", backfill_state.pending_runs[0]) + def test_skips_completed_backfill(self): class WindowClient(FakeClient): def __init__(self): diff --git a/github-actions-queue/README.md b/github-actions-queue/README.md index 348ad26f52a5..56df60460d57 100644 --- a/github-actions-queue/README.md +++ b/github-actions-queue/README.md @@ -120,10 +120,22 @@ within the workflow timeout. Later runs use the normal 40-minute limit. Runs that have not reached a terminal state are saved in `state.json`. Later collections revisit them and emit their jobs only after every returned job is -terminal. Job listing uses `filter=all`, so actual executions from every +terminal. A job reported as `in_progress` is also terminal when GitHub supplies +a completion timestamp and a terminal conclusion. This handles stale status +values without finalizing jobs that still lack completion evidence. Records +retain GitHub's reported job status and conclusion. +Job listing uses `filter=all`, so actual executions from every available attempt are retained when the run is finalized. Carry-forward clones created by partial reruns are not stored. +Each live or backfill collection selects up to 25 saved pending runs before +collecting windows. Five slots are reserved for unresolved API failures and +20 for other unfinished runs; unused slots are shared. Both groups use the +oldest last-attempt time, falling back to creation time for unattempted runs. +Selected runs are processed in that same chronological order, within the +request, runtime, and REST quota limits. Backfill retries run only after live +collection catches up without pausing and leaves a window budget. + The collector tracks the installation's REST quota. If it exhausts the quota, it commits the partial checkpoint and resumes during the next scheduled run. @@ -134,7 +146,10 @@ per-file counts during validation, then records completion under GitHub limits a filtered workflow-run search to 1,000 results. The collector splits busy one-hour windows into smaller ranges until each result can be -paginated safely. Jobs are paginated at 100 per request. +paginated safely. Repository and workflow-run pages contain up to 100 results. +Job pages contain up to 50 results to reduce GitHub jobs-API server errors on +large pages. The collector checks that all pages yield the reported number of +unique jobs before finalizing a run. ## Data layout @@ -158,7 +173,7 @@ reporting data. Each job line contains: | `event`, `from_fork` | Trigger and whether the head repository was a fork. `from_fork` is null when GitHub no longer provides the head repository. | | `head_branch`, `head_sha` | Source revision. | | `job_id`, `job_name` | Individual job identity. Matrix values normally appear in `job_name`. | -| `job_status`, `job_conclusion` | Terminal state and result. | +| `job_status`, `job_conclusion` | GitHub's reported state and result. A terminal job can retain a stale `in_progress` status. | | `job_created_at`, `job_started_at`, `job_completed_at` | GitHub job timestamps. | | `queue_seconds` | Start minus creation time. | | `runner_assigned` | Always true for retained queue records. | @@ -204,8 +219,11 @@ Failed job lookups are retained in `state.json` with their failure count, last error, and last attempt time. Backfill failures are retained in `backfill-state.json` the same way. Later collections retry them while continuing to discover other runs. The workflow emits a warning as long as any -failed lookup remains unresolved; failures are never converted into successful -empty records. +failed lookup remains unresolved. A successful job listing clears its last +error even if jobs are still unfinished, while retaining the historical failure +count. The summary's `pending_run_failures` counts unresolved errors, not errors +encountered only in the current collection. Failures are never converted into +successful empty records. GitHub announced that organization retention settings will cover workflow runs created on or after October 1, 2026. The change is not retroactive. The January