This is an automated email from the ASF dual-hosted git repository.
shahar1 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new 8693c0854fe Retry fork CI job listings in smaller pages (#76327)
8693c0854fe is described below
commit 8693c0854fe24b94c5ff9f9dc0a675c1315cd165
Author: Jarek Potiuk <[email protected]>
AuthorDate: Tue Nov 3 13:57:21 2026 +0100
Retry fork CI job listings in smaller pages (#76327)
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
