This is an automated email from the ASF dual-hosted git repository. potiuk pushed a commit to branch fork-ci-jobs-smaller-pages in repository https://gitbox.apache.org/repos/asf/airflow.git
commit 7b80c0850c8cf6763d66a4205354f0e0fd720357 Author: Jarek Potiuk <[email protected]> AuthorDate: Mon Nov 2 21:09:28 2026 +0100 Retry fork CI job listings in smaller pages GitHub answers HTTP 502 on every attempt when it lists 100 jobs per page of some large fork runs that were re-run, while smaller pages of the same run list fine. The fork CI monitor then never read the re-run's jobs, so job statuses of the failed first attempt stayed red on the pull request after the re-run passed, as on #74696. Generated-by: Claude Opus 5 --- .../src/airflow_breeze/utils/fork_ci_github.py | 59 ++++++++++++++---- .../src/airflow_breeze/utils/fork_ci_reconcile.py | 12 +--- dev/breeze/tests/test_fork_ci_github.py | 69 ++++++++++++++++++++++ dev/breeze/tests/test_fork_ci_reconcile.py | 15 ++--- 4 files changed, 126 insertions(+), 29 deletions(-) diff --git a/dev/breeze/src/airflow_breeze/utils/fork_ci_github.py b/dev/breeze/src/airflow_breeze/utils/fork_ci_github.py index bc763c7e546..9d84bc89f1c 100644 --- a/dev/breeze/src/airflow_breeze/utils/fork_ci_github.py +++ b/dev/breeze/src/airflow_breeze/utils/fork_ci_github.py @@ -45,6 +45,11 @@ SERVER_ERROR = re.compile(r"\bHTTP 5\d\d\b|connection reset by peer|unexpected E # Network failures before the request reached GitHub, so repeating it is safe for any call. CONNECTION_ERROR = re.compile(r"TLS handshake timeout|dial tcp") SERVER_ERROR_WAIT_SECONDS = 10 +# GitHub answers HTTP 502 on every attempt when it lists 100 jobs per page of some runs with a few +# hundred jobs, re-run more than once, while it lists smaller pages of the same run. A smaller size is +# tried once the previous one keeps failing with server errors. +JOB_PAGE_SIZES = (100, 30, 15, 5) +JOB_PAGE_SIZE_SERVER_ERROR_ATTEMPTS = 2 # A hung `gh` would otherwise hold the job until its timeout-minutes. GH_TIMEOUT_SECONDS = 120 @@ -69,6 +74,7 @@ class GitHubClient: *, allow_error_output: bool = False, retry_server_errors: bool = True, + max_server_error_attempts: int = MAX_ATTEMPTS, ) -> Any: """ :param allow_error_output: return the JSON output of a failed call when it has any. ``gh api @@ -77,6 +83,8 @@ class GitHubClient: :param retry_server_errors: repeat a call that failed with a server error. Off for a call that may have taken effect anyway and must not be repeated, such as creating a comment. A call that failed to connect is repeated either way. + :param max_server_error_attempts: how many times the call is made before giving up on server + errors; rate limits and connection errors are retried up to ``MAX_ATTEMPTS`` times either way. """ payload = json.dumps(body) if body is not None else None rate_limited = False @@ -98,12 +106,10 @@ class GitHubClient: rate_limited = any(marker in result.stderr.lower() for marker in RATE_LIMIT_MARKERS) self.rate_limited += rate_limited connection_error = bool(CONNECTION_ERROR.search(result.stderr)) - if ( - rate_limited - or connection_error - or (retry_server_errors and SERVER_ERROR.search(result.stderr)) - ): - if attempt == MAX_ATTEMPTS - 1: + server_error = not rate_limited and not connection_error and SERVER_ERROR.search(result.stderr) + if rate_limited or connection_error or (retry_server_errors and server_error): + last_attempt = max_server_error_attempts if server_error else MAX_ATTEMPTS + if attempt >= last_attempt - 1: break reason = ( "rate limit" @@ -123,9 +129,9 @@ class GitHubClient: pass raise GitHubError(f"gh {' '.join(args)} failed: {result.stderr.strip()}") if rate_limited: - raise GitHubError(f"gh {' '.join(args)} still rate limited after {MAX_ATTEMPTS} attempts") + raise GitHubError(f"gh {' '.join(args)} still rate limited after {attempt + 1} attempts") raise GitHubError( - f"gh {' '.join(args)} still failing after {MAX_ATTEMPTS} attempts: {result.stderr.strip()}" + f"gh {' '.join(args)} still failing after {attempt + 1} attempts: {result.stderr.strip()}" ) def graphql( @@ -149,20 +155,53 @@ class GitHubClient: console_print(f"[warning]Partial GraphQL result: {response['errors']}[/]") return response["data"] - def rest_get(self, path: str, *, paginate: bool = False, items_key: str | None = None) -> Any: + def rest_get( + self, + path: str, + *, + paginate: bool = False, + items_key: str | None = None, + max_server_error_attempts: int = MAX_ATTEMPTS, + ) -> Any: """ GET a REST path. :param paginate: follow every page and return the concatenated items. :param items_key: for endpoints whose pages are objects, the key holding the items (``workflow_runs``, ``jobs``, ``files``); pages are plain lists otherwise. + :param max_server_error_attempts: how many times the call is made before giving up on server + errors. """ args = ["api", path, *(["--paginate", "--slurp"] if paginate else [])] - result = self._gh(args, None, "rest") + result = self._gh(args, None, "rest", max_server_error_attempts=max_server_error_attempts) if paginate: return [item for page in result for item in (page[items_key] if items_key else page)] return result + def get_run_jobs(self, repo: str, run_id: int) -> list[dict[str, Any]]: + """The jobs of a workflow run's latest attempt, in smaller pages while GitHub fails to list them.""" + + def get_jobs(page_size: int, max_server_error_attempts: int) -> list[dict[str, Any]]: + return self.rest_get( + f"repos/{repo}/actions/runs/{run_id}/jobs?filter=latest&per_page={page_size}", + paginate=True, + items_key="jobs", + max_server_error_attempts=max_server_error_attempts, + ) + + *larger, smallest = JOB_PAGE_SIZES + for page_size in larger: + try: + return get_jobs(page_size, JOB_PAGE_SIZE_SERVER_ERROR_ATTEMPTS) + except GitHubError as e: + if not SERVER_ERROR.search(str(e)): + raise + console_print( + f"[warning]Listing the jobs of {repo} run {run_id} {page_size} per page keeps " + "failing, trying smaller pages[/]" + ) + return get_jobs(smallest, MAX_ATTEMPTS) + def rest_get_pages(self, path: str) -> list[Any]: """GET every page of a paginated REST path, keeping page-level fields such as ``total_count``.""" return self._gh(["api", path, "--paginate", "--slurp"], None, "rest") diff --git a/dev/breeze/src/airflow_breeze/utils/fork_ci_reconcile.py b/dev/breeze/src/airflow_breeze/utils/fork_ci_reconcile.py index 80dd7242a3e..d642b573e5f 100644 --- a/dev/breeze/src/airflow_breeze/utils/fork_ci_reconcile.py +++ b/dev/breeze/src/airflow_breeze/utils/fork_ci_reconcile.py @@ -1395,11 +1395,7 @@ class JobStatusReporter: if self.remaining == 0: self._defer(pr, fork) return None - jobs = self._client.rest_get( - f"repos/{pr.head_repo}/actions/runs/{fork.run_id}/jobs?filter=latest&per_page=100", - paginate=True, - items_key="jobs", - ) + jobs = self._client.get_run_jobs(pr.head_repo, fork.run_id) # By the case-folded context, as GitHub compares them. current = { status["context"].casefold(): ( @@ -1552,11 +1548,7 @@ class BypassFinder: def _get_jobs(self, run: WorkflowRun) -> list[dict[str, Any]]: if run.id not in self._jobs: - self._jobs[run.id] = self._client.rest_get( - f"repos/{self._repo}/actions/runs/{run.id}/jobs?filter=latest&per_page=100", - paginate=True, - items_key="jobs", - ) + self._jobs[run.id] = self._client.get_run_jobs(self._repo, run.id) return self._jobs[run.id] def _did_run_project_ci(self, run: WorkflowRun) -> bool: diff --git a/dev/breeze/tests/test_fork_ci_github.py b/dev/breeze/tests/test_fork_ci_github.py index 25ae3f105c6..ab497dd3b34 100644 --- a/dev/breeze/tests/test_fork_ci_github.py +++ b/dev/breeze/tests/test_fork_ci_github.py @@ -238,3 +238,72 @@ def test_set_status_context(mock_run, kwargs, context): GitHubClient().set_status(1, "abc", "success", "https://x";, "desc", **kwargs) assert json.loads(mock_run.call_args.kwargs["input"])["context"] == context + + +def _jobs_page(*names: str) -> str: + return json.dumps([{"total_count": len(names), "jobs": [{"name": name} for name in names]}]) + + +def _jobs_path(page_size: int) -> str: + return f"repos/someone/airflow/actions/runs/7/jobs?filter=latest&per_page={page_size}" + + +BAD_GATEWAY = "gh: Server Error (HTTP 502)" + + [email protected]("airflow_breeze.utils.fork_ci_github.time.sleep", autospec=True) [email protected]("airflow_breeze.utils.fork_ci_github.subprocess.run", autospec=True) +def test_lists_run_jobs_in_smaller_pages_while_github_fails_to_list_them(mock_run, mock_sleep): + mock_run.side_effect = [ + *[_completed(1, stderr=BAD_GATEWAY)] * 4, + _completed(0, stdout=_jobs_page("Tests", "Docs")), + ] + + jobs = GitHubClient().get_run_jobs("someone/airflow", 7) + + assert jobs == [{"name": "Tests"}, {"name": "Docs"}] + assert [c.args[0][2] for c in mock_run.call_args_list] == [ + _jobs_path(100), + _jobs_path(100), + _jobs_path(30), + _jobs_path(30), + _jobs_path(15), + ] + + [email protected]("airflow_breeze.utils.fork_ci_github.time.sleep", autospec=True) [email protected]("airflow_breeze.utils.fork_ci_github.subprocess.run", autospec=True) +def test_gives_up_listing_run_jobs_when_the_smallest_pages_keep_failing(mock_run, mock_sleep): + mock_run.return_value = _completed(1, stderr=BAD_GATEWAY) + + with pytest.raises(GitHubError, match=r"per_page=5 --paginate --slurp still failing after 4 attempts"): + GitHubClient().get_run_jobs("someone/airflow", 7) + assert [c.args[0][2] for c in mock_run.call_args_list] == [ + *[_jobs_path(100)] * 2, + *[_jobs_path(30)] * 2, + *[_jobs_path(15)] * 2, + *[_jobs_path(5)] * 4, + ] + + [email protected]("airflow_breeze.utils.fork_ci_github.time.sleep", autospec=True) [email protected]("airflow_breeze.utils.fork_ci_github.subprocess.run", autospec=True) +def test_listing_run_jobs_retries_a_rate_limit_as_often_as_any_call(mock_run, mock_sleep): + mock_run.side_effect = [ + *[_completed(1, stderr="API rate limit exceeded")] * 3, + _completed(0, stdout=_jobs_page("Tests")), + ] + + assert GitHubClient().get_run_jobs("someone/airflow", 7) == [{"name": "Tests"}] + assert [c.args[0][2] for c in mock_run.call_args_list] == [_jobs_path(100)] * 4 + assert [c.args[0] for c in mock_sleep.call_args_list] == [30, 60, 120] + + [email protected]("airflow_breeze.utils.fork_ci_github.time.sleep", autospec=True) [email protected]("airflow_breeze.utils.fork_ci_github.subprocess.run", autospec=True) +def test_does_not_shrink_run_job_pages_for_errors_other_than_server_errors(mock_run, mock_sleep): + mock_run.return_value = _completed(1, stderr="gh: Not Found (HTTP 404)") + + with pytest.raises(GitHubError, match="HTTP 404"): + GitHubClient().get_run_jobs("someone/airflow", 7) + mock_run.assert_called_once() diff --git a/dev/breeze/tests/test_fork_ci_reconcile.py b/dev/breeze/tests/test_fork_ci_reconcile.py index 417a08540af..41c6b970794 100644 --- a/dev/breeze/tests/test_fork_ci_reconcile.py +++ b/dev/breeze/tests/test_fork_ci_reconcile.py @@ -933,8 +933,6 @@ def _finder( return [{"total_count": count, "workflow_runs": event_runs}] def rest_get(path, paginate=False, items_key=None): - if path.endswith("jobs?filter=latest&per_page=100"): - return list(jobs) if "/compare/" in path: assert not paginate, "the compare API returns files only on its first page" return {"files": [{"filename": name} for name in compare_files]} @@ -944,6 +942,7 @@ def _finder( client.rest_get_pages.side_effect = rest_get_pages client.rest_get.side_effect = rest_get + client.get_run_jobs.return_value = list(jobs) return BypassFinder(client, repo=REPO, now=NOW, base_triggers=BASE_TRIGGERS, is_gated_base=is_gated_base) @@ -1130,9 +1129,7 @@ def test_finder_reports_a_run_another_pull_request_may_own(client, other_pulls): def test_finder_catches_a_workflow_moved_away(client): finder = _finder(client, [_run(7)], jobs=TAMPERED_JOBS) client.rest_get.side_effect = lambda path, paginate=False, items_key=None: ( - TAMPERED_JOBS - if path.endswith("jobs?filter=latest&per_page=100") - else [{"number": 42, "closed_at": None}] + [{"number": 42, "closed_at": None}] if "/pulls?head=" in path else {"files": [{"filename": "docs/x.yml", "previous_filename": ".github/workflows/x.yml"}]} ) @@ -2087,13 +2084,12 @@ def test_run_reconcile_deletes_gated_runs(mock_classify, client, tmp_path, state def rest_get(path, paginate=False, items_key=None): if path.endswith("/actions/workflows/ci-amd.yml"): return {"state": "active"} - if path.endswith("jobs?filter=latest&per_page=100"): - return GATED_JOBS if "/compare/" in path: return {"files": [{"filename": "README.md"}]} raise AssertionError(path) client.rest_get.side_effect = rest_get + client.get_run_jobs.return_value = GATED_JOBS (tmp_path / "ci-amd.yml").write_text("on: [pull_request]\n") run_reconcile( @@ -2413,13 +2409,14 @@ COMPLETED_RUN = ForkObservation( def _reporter_client(client, jobs, current=()): def rest_get(path, paginate=False, items_key=None): - if path.startswith("repos/someone/airflow/actions/runs/1/jobs"): - return jobs if path.startswith(f"repos/{REPO}/commits/abc123/status"): return list(current) raise AssertionError(path) client.rest_get.side_effect = rest_get + client.get_run_jobs.side_effect = lambda repo, run_id: ( + jobs if (repo, run_id) == ("someone/airflow", 1) else pytest.fail(f"{repo} run {run_id}") + ) return client
