This is an automated email from the ASF dual-hosted git repository.
potiuk 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 468ab0a5834 Spend fewer GitHub writes on fork CI job statuses and
watch the limits (#76154)
468ab0a5834 is described below
commit 468ab0a5834c8ee676f99e0be8bd90b440423862
Author: Jarek Potiuk <[email protected]>
AuthorDate: Sat Oct 31 18:11:03 2026 +0100
Spend fewer GitHub writes on fork CI job statuses and watch the limits
(#76154)
* Spend fewer GitHub writes on fork CI job statuses and watch the limits
The fork CI monitor's per-job commit statuses are its largest use of
the GITHUB_TOKEN limits that every workflow of the repository shares
(500 content-creating requests an hour, 80 a minute). During the pilot
they already fill the monitor's per-run budget whenever a couple of
full-matrix fork runs are active, which would leave little headroom for
other workflows once fork CI is enabled for every contributor.
A "running" status for each job doubled the writes while saying little:
the monitor only runs every 15 minutes, so it lagged behind the fork's
own run page, and the aggregate Fork CI status already shows the run as
running and links it.
Until now the only sign of hitting the limits was failures, so each run
now logs the quota left and how many calls GitHub rate-limited, and
Slack hears about rate limiting that persists.
Generated-by: Claude Opus 5
* Report pull requests whose fork job statuses lag far behind
Generated-by: Claude Opus 5
---
dev/AGENTS.md | 7 +-
.../doc/ci/08_fork_ci_for_external_contributors.md | 58 +++++----
.../src/airflow_breeze/utils/fork_ci_github.py | 3 +
.../src/airflow_breeze/utils/fork_ci_messages.py | 7 +-
.../src/airflow_breeze/utils/fork_ci_reconcile.py | 105 +++++++++++++---
dev/breeze/tests/test_fork_ci_github.py | 6 +-
dev/breeze/tests/test_fork_ci_reconcile.py | 133 +++++++++++++++++++--
7 files changed, 259 insertions(+), 60 deletions(-)
diff --git a/dev/AGENTS.md b/dev/AGENTS.md
index 24267879fdb..f9e999acfc4 100644
--- a/dev/AGENTS.md
+++ b/dev/AGENTS.md
@@ -55,8 +55,9 @@ workflow of `apache/airflow` shares (see "Write budget" in
If the budget is exceeded, turn the job statuses off: set `job_statuses =
false` in
`.github/fork-ci-gate.toml`. Signs that it is exceeded are the monitor's Slack
note that job
statuses "have not fit in the budget" (posted only after three runs in a row
deferred them)
-staying for hours, rate-limit retries ("GitHub rate limit,
-retrying") or "still rate limited" errors in the monitor's run log or its
Slack messages, or other
-workflows of the repository failing on secondary rate limits. The aggregate
`Fork CI / Tests (AMD)`
+staying for hours, pull requests listed because their fork run finished more
than an hour ago with
+job statuses still not posted, its Slack note that GitHub rate-limited it in
two runs in a row, the
+"Rate limits left" line of its run log approaching zero, or other workflows of
the repository
+failing on secondary rate limits. The aggregate `Fork CI / Tests (AMD)`
status and `CI passed` are posted either way, so turning job statuses off
loses detail, not gating.
Lowering `MAX_JOB_STATUSES_PER_RUN` in `fork_ci_reconcile.py` is the softer
alternative.
diff --git a/dev/breeze/doc/ci/08_fork_ci_for_external_contributors.md
b/dev/breeze/doc/ci/08_fork_ci_for_external_contributors.md
index 51b595aa268..181180b5516 100644
--- a/dev/breeze/doc/ci/08_fork_ci_for_external_contributors.md
+++ b/dev/breeze/doc/ci/08_fork_ci_for_external_contributors.md
@@ -805,15 +805,18 @@ suites); its jobs come from `GET
/repos/<fork>/actions/runs/<id>/jobs?filter=lat
the prefix is what keeps a job from posting `Fork CI / Tests (AMD)`, from
which the reconciler
reads its decisions back, or `CI passed`, which the merge ruleset requires.
Jobs that share a
context, which GitHub compares case-insensitively, are reported as the worst
of them.
-- A queued or running job is `pending` ("Running in the fork"), a successful
one `success`, and a
- failed, timed-out, cancelled or unstartable one `failure`. Skipped and
neutral jobs, and the jobs
- every run has (`Fork CI gate`, `Check if the run is needed`, `Platform: AMD`
and `CI passed`), are
- not reported. Queued and running share one status so that starting a job
costs no write.
+- Only finished jobs are reported: a successful one as `success`, and a
failed, timed-out,
+ cancelled or unstartable one as `failure`. Queued and running jobs are not,
so each job costs one
+ write instead of two; `Fork CI / Tests (AMD)` shows the run as running,
linking it, and with a
+ 15-minute cadence a per-job "running" status was behind the fork's own run
page anyway. A job of
+ a re-run in the fork keeps the result of the earlier attempt until it
finishes. Skipped and
+ neutral jobs, and the jobs every run has (`Fork CI gate`, `Check if the run
is needed`,
+ `Platform: AMD` and `CI passed`), are not reported.
- The desired statuses are compared with the head commit's current ones
(`GET /repos/apache/airflow/commits/<sha>/status`), and only differences are
posted, failures
first. GraphQL has no mutation for commit statuses (`createCheckRun` exists,
but see above for
why check runs do not fit), so each one is a REST call, and they count
towards GitHub's limits on
- content-creating requests: a reconciler run posts at most
`MAX_JOB_STATUSES_PER_RUN` (60) of
+ content-creating requests: a reconciler run posts at most
`MAX_JOB_STATUSES_PER_RUN` (40) of
them, each write charged as it is made. What does not fit is posted by the
next run, and once
the budget is spent no further pull request's jobs or statuses are read in
that run. One fork run
of the full matrix (about 130 jobs) needs more than a run's budget on its
own, which the next
@@ -821,7 +824,11 @@ suites); its jobs come from `GET
/repos/<fork>/actions/runs/<id>/jobs?filter=lat
requests, and only when three runs in a row deferred, about 45 minutes of
the budget not
keeping up, is a note added to the Slack message. The count is kept in the
record the monitor
carries between runs (`--reported-file`); without a readable record the note
is added at once.
- Its wording is fixed, so it is posted once while deferrals persist.
+ Its wording is fixed, so it is posted once while deferrals persist. A pull
request whose fork run
+ finished more than an hour ago (the `updatedAt` of its completed check
suite) and still has
+ statuses waiting is listed on its own as an attention item, so a delay that
a busy backlog keeps
+ pushing back shows which pull request it holds up; the item names the run,
so it is posted once
+ for that run.
- Commit statuses cannot be deleted and have no neutral state. When a run has
completed, a job
context posted for an earlier run of the same commit that the completed run
does not have, or
skipped, is overwritten with `success` and the description "Not in the
latest fork run - see
@@ -839,9 +846,14 @@ on `Fork CI / Tests (AMD)`; a job status never changes
either.
`job_statuses = false` in `.github/fork-ci-gate.toml` turns them off; the
reconciler then posts
only `Fork CI / Tests (AMD)` and `CI passed`, as before. Statuses already
posted stay on their
-commits. Turn them off when the fork CI monitor's runs hit GitHub's rate
limits (rate-limit
-retries in the run log, or "still rate limited" errors in Slack);
`dev/AGENTS.md` records this for
-agents.
+commits. Turn them off when the fork CI monitor's runs hit GitHub's rate
limits; `dev/AGENTS.md`
+records this for agents.
+
+Each run logs what is left of the token's REST and GraphQL quotas (`GET
/rate_limit`, which does
+not count against them) and how many of its calls GitHub rate-limited; the
client retries those with
+backoff. Rate limiting in two runs in a row adds a note to the Slack message,
with fixed wording so
+it is posted once while it persists; the count is kept in the same record as
that of deferred job
+statuses.
#### Write budget
@@ -857,25 +869,25 @@ jobs whose conclusion is not `skipped`, minus the four
jobs that are not reporte
| CI change (full tests) |
[40207881565](https://github.com/apache/airflow/actions/runs/40207881565) | 131
| 127 |
`apache/airflow` had 1898 `Tests (AMD)` pull request runs from 2026-10-18 to
2026-10-24, about 270 a
-day or 11 an hour. Each reported job costs at most two writes (running, then
its result) and one
-when it starts and finishes between two reconciler runs; a fork run of 30-60
minutes spans two to
-four of them, so most jobs cost two.
+day or 11 an hour. Each reported job costs one write, its result.
The limits that matter are GitHub's documented ones for the `GITHUB_TOKEN` of
`apache/airflow`,
shared by every workflow of the repository: 1000 REST requests an hour, and 80
content-creating
requests a minute and 500 an hour. Assuming about 35 reported jobs for an
average push:
-- **Contributor pilot**, a few gated pushes a day: about 70 writes a push,
under 20 an hour, under
- 5% of the content-creating limit.
+- **Contributor pilot**, a few gated pushes a day: about 35 writes a push,
under 10 an hour, under
+ 2% of the content-creating limit. A single full-matrix run (about 127 jobs)
still spans a few
+ monitor runs.
- **Gate enabled**, assuming about half of the pull request runs come from
gated pull requests
- (about 5-6 pushes an hour): about 400 writes an hour without a cap, 80% of
the content-creating
- limit and more at peaks. The cap of 60 writes a run, four runs an hour,
holds job statuses to at
- most 240 an hour: 48% of the content-creating limit and 24% of the REST
limit, plus two reads per
- pull request with a run in progress. Job statuses then lag behind at busy
times, the aggregate
- status does not. The `CI run started` statuses of pull requests whose CI
runs in
- `apache/airflow` come out of the same 60 writes a run, first and at most 30
of them (see "The
- `CI run started` check"), so they do not raise the total; at about 11 pull
request runs an hour
- they take about 3 writes a run.
+ (about 5-6 pushes an hour): about 200 writes an hour without a cap, 40% of
the content-creating
+ limit and more at peaks. The cap of 40 writes a run, four runs an hour,
holds job statuses to at
+ most 160 an hour: 32% of the content-creating limit, leaving the rest to the
reconciler's other
+ writes (labels, comments, the aggregate statuses) and to the other workflows
that share it. Each
+ run posts at most 40 statuses at about one a second, well under the
80-a-minute limit. Job
+ statuses then lag behind at busy times, the aggregate status does not. The
`CI run started`
+ statuses of pull requests whose CI runs in `apache/airflow` come out of the
same 40 writes a run,
+ first and at most 20 of them (see "The `CI run started` check"), so they do
not raise the total;
+ at about 11 pull request runs an hour they take about 3 writes a run.
These are estimates: the share of gated runs, the jobs per push and the pushes
per pull request
are assumptions, not measurements. Watch the monitor's call counts in its run
log after enabling
@@ -936,7 +948,7 @@ with state `success` linking the run:
Folding the check into the scheduled monitor costs no runner per push, at the
price of a pull
request waiting for the next monitor run before it can merge, which a CI run
takes longer than
anyway. Each new head commit needs one write, taken first out of the per-run
budget of job
-statuses (see "Job statuses"), since merging waits for it; at most 30 such
writes per run, so job
+statuses (see "Job statuses"), since merging waits for it; at most 20 such
writes per run, so job
statuses keep at least half of the budget, and after the check became required
the open pull
requests are caught up a batch at a time. A release branch has no monitor of
its own, but `main`'s monitor covers pull
requests to every base branch.
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 77a7ef91011..bc763c7e546 100644
--- a/dev/breeze/src/airflow_breeze/utils/fork_ci_github.py
+++ b/dev/breeze/src/airflow_breeze/utils/fork_ci_github.py
@@ -58,6 +58,8 @@ class GitHubClient:
self.repo = repo
self.dry_run = dry_run
self.calls: Counter[str] = Counter()
+ # Calls GitHub answered with a rate limit, each retried or given up on.
+ self.rate_limited = 0
def _gh(
self,
@@ -94,6 +96,7 @@ class GitHubClient:
if result.returncode == 0:
return json.loads(result.stdout) if result.stdout.strip() else
None
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
diff --git a/dev/breeze/src/airflow_breeze/utils/fork_ci_messages.py
b/dev/breeze/src/airflow_breeze/utils/fork_ci_messages.py
index 0814021495b..6fb8d04aac5 100644
--- a/dev/breeze/src/airflow_breeze/utils/fork_ci_messages.py
+++ b/dev/breeze/src/airflow_breeze/utils/fork_ci_messages.py
@@ -304,11 +304,10 @@ _STATUS_DESCRIPTIONS = {
"gate_modified": "Pull request changes the fork CI gate - see the comment",
}
-# (commit status state, description) of a fork job, by its status and
conclusion. A queued job is
-# reported like a running one: each change is one more write, and commit
statuses can only be posted
-# one at a time (GraphQL has no mutation for them).
+# (commit status state, description) of a finished fork job, by its
conclusion. Queued and running
+# jobs are not reported: each status is one write, commit statuses can only be
posted one at a time
+# (GraphQL has no mutation for them), and `Fork CI / Tests (AMD)` already
shows the run as running.
JOB_STATUSES = {
- "running": ("pending", "Running in the fork - see Details"),
"success": ("success", "Passed in the fork - see Details"),
"failure": ("failure", "Failed in the fork - see Details"),
"timed_out": ("failure", "Timed out in the fork - see Details"),
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 686ccc44136..d940d8a21ce 100644
--- a/dev/breeze/src/airflow_breeze/utils/fork_ci_reconcile.py
+++ b/dev/breeze/src/airflow_breeze/utils/fork_ci_reconcile.py
@@ -155,18 +155,29 @@ JOB_STATE_RANKS = {"failure": 0, "pending": 1, "success":
2}
# Commit statuses count towards GitHub's limit of 500 content-creating
requests per hour, shared by
# every workflow of the repository. At four runs an hour this keeps job
statuses to at most about
# half of it; the rest follow on the next run, failures first. See "Job
statuses" in the design doc.
-MAX_JOB_STATUSES_PER_RUN = 60
+MAX_JOB_STATUSES_PER_RUN = 40
# `CI run started` statuses of pull requests whose CI runs in apache/airflow
come out of that budget
# first: one write per new head commit, which merging waits for, while job
statuses are informational.
# At most this many, so job statuses keep the rest; the first runs after the
check became required
# catch up on the open pull requests a batch at a time, most recently updated
first.
-MAX_CI_RUN_STARTED_PER_RUN = 30
+MAX_CI_RUN_STARTED_PER_RUN = 20
# The GitHub Actions app, whose check suites are the workflow runs of a commit.
GITHUB_ACTIONS_APP_ID = 15368
# A single large fork run can need more writes than one run's budget, which
the next run catches up
# on, so deferring is only reported when it lasts this many consecutive runs:
then the budget does
# not keep up with the pull requests.
DEFERRED_RUNS_BEFORE_ALERT = 3
+# GitHub answering with rate limits in this many consecutive runs is reported:
a single burst is
+# retried, while a persisting one means the repository's shared limits are
close to exhausted.
+RATE_LIMITED_RUNS_BEFORE_ALERT = 2
+# A pull request whose fork run finished this long ago with job statuses still
not all posted is
+# reported by itself: deferring is expected to catch up within a few runs, not
to leave one behind.
+MIRROR_DELAY_BEFORE_ALERT = timedelta(hours=1)
+RATE_LIMITED_NOTE = (
+ f"GitHub rate-limited the fork CI monitor in
{RATE_LIMITED_RUNS_BEFORE_ALERT} runs in a row; the "
+ "repository's shared GITHUB_TOKEN limits are close to exhausted. Consider
`job_statuses = false` in "
+ ".github/fork-ci-gate.toml (see dev/AGENTS.md)"
+)
# Fixed text, so the note is posted to Slack once while deferrals persist; the
run log lists the pull
# requests.
JOB_STATUSES_DEFERRED_NOTE = (
@@ -277,6 +288,8 @@ class ForkObservation:
run_url: str | None = None
run_id: int | None = None
run_attempt: int | None = None
+ # When the fork's check suite last changed, for a completed one when the
run finished.
+ completed_at: datetime | None = None
workflow_state: str | None = None
error: str | None = None
@@ -1248,7 +1261,7 @@ def fetch_fork_observations(
declarations.append(f"$o{index}: String!, $n{index}: String!,
$s{index}: GitObjectID!")
fields.append(
f"pr{index}: repository(owner: $o{index}, name: $n{index}) {{
object(oid: $s{index}) {{ "
- "... on Commit { checkSuites(first: 30) { nodes { createdAt
status conclusion "
+ "... on Commit { checkSuites(first: 30) { nodes { createdAt
updatedAt status conclusion "
"workflowRun { databaseId runAttempt url event workflow {
resourcePath } } } } } } }"
)
variables.update({f"o{index}": owner, f"n{index}": name,
f"s{index}": pr.head_sha})
@@ -1279,24 +1292,27 @@ def fetch_fork_observations(
run_url=latest["workflowRun"]["url"],
run_id=latest["workflowRun"].get("databaseId"),
run_attempt=latest["workflowRun"].get("runAttempt"),
+ completed_at=(
+ parse_github_datetime(latest.get("updatedAt") or "")
+ if latest["status"] == "COMPLETED"
+ else None
+ ),
)
return observations
def build_job_statuses(jobs: Iterable[dict[str, Any]]) -> dict[str, tuple[str,
str, str]]:
- """The ``(state, target URL, description)`` to report for each queued or
started fork job, by context."""
+ """The ``(state, target URL, description)`` to report for each finished
fork job, by context."""
statuses: dict[str, tuple[str, str, str]] = {}
# GitHub compares contexts case-insensitively, so differently capitalised
names are one context.
contexts: dict[str, str] = {}
for job in jobs:
if job["name"] in UNREPORTED_JOB_NAMES:
continue
- if job.get("status") in RUNNING_JOB_STATUSES:
- outcome = "running"
- else:
- outcome = job.get("conclusion") or ""
+ outcome = "" if job.get("status") in RUNNING_JOB_STATUSES else
job.get("conclusion") or ""
if outcome not in JOB_STATUSES:
- # Skipped and neutral jobs did no work worth a check on the pull
request.
+ # Unfinished jobs are left to `Fork CI / Tests (AMD)`; skipped and
neutral ones did no work
+ # worth a check on the pull request.
continue
state, description = JOB_STATUSES[outcome]
context = contexts.setdefault(
@@ -1312,11 +1328,38 @@ def build_job_statuses(jobs: Iterable[dict[str, Any]])
-> dict[str, tuple[str, s
class JobStatusReporter:
"""Reports each job of a pull request's fork run as a commit status,
within a per-run budget."""
- def __init__(self, client: GitHubClient, *, repo: str, budget: int =
MAX_JOB_STATUSES_PER_RUN):
+ def __init__(
+ self,
+ client: GitHubClient,
+ *,
+ repo: str,
+ budget: int = MAX_JOB_STATUSES_PER_RUN,
+ now: datetime | None = None,
+ ):
+ """:param now: when given, deferring the statuses of a run that
finished long ago is reported."""
self._client = client
self._repo = repo
+ self._now = now
self.remaining = budget
self.deferred: list[int] = []
+ # (pull request, attention item) for runs whose statuses are deferred
long after they finished.
+ self.delayed: list[tuple[int, str]] = []
+
+ def _defer(self, pr: PullRequestSnapshot, fork: ForkObservation) -> None:
+ self.deferred.append(pr.number)
+ if (
+ self._now is not None
+ and fork.completed_at is not None
+ and self._now - fork.completed_at >= MIRROR_DELAY_BEFORE_ALERT
+ ):
+ # Fixed text per run, so it is posted to Slack once while the
delay lasts.
+ self.delayed.append(
+ (
+ pr.number,
+ f"job statuses of fork run {fork.run_url} are still not
all posted more than "
+ f"{MIRROR_DELAY_BEFORE_ALERT} after it finished",
+ )
+ )
def report(self, pr: PullRequestSnapshot, fork: ForkObservation) -> str |
None:
"""
@@ -1340,7 +1383,7 @@ class JobStatusReporter:
# Without writes left, the reads are spent for nothing: a backlog of
hundreds of pull requests
# would otherwise cost two reads each on every run.
if self.remaining == 0:
- self.deferred.append(pr.number)
+ 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",
@@ -1376,7 +1419,7 @@ class JobStatusReporter:
)
for context, (state, target_url, description) in changes:
if self.remaining == 0:
- self.deferred.append(pr.number)
+ self._defer(pr, fork)
return None
self._client.set_status(pr.number, pr.head_sha, state, target_url,
description, context=context)
# Charged per write, so writes made before a later one fails still
count.
@@ -2041,6 +2084,7 @@ def drop_reported(
notes: list[str],
record: bool = True,
deferred_runs: int = 0,
+ rate_limited_runs: int = 0,
) -> tuple[list[tuple[int, str]], list[tuple[int, str]], list[str]]:
"""
Leave out the items the previous run already reported, and record this
run's items.
@@ -2049,12 +2093,14 @@ def drop_reported(
again. A missing or unreadable file reports everything.
:param deferred_runs: the consecutive runs that deferred job statuses,
recorded for the next run.
+ :param rate_limited_runs: the consecutive runs GitHub rate-limited,
recorded for the next run.
"""
current: dict[str, Any] = {
"unknown": [list(item) for item in unknown],
"attention": [list(item) for item in attention],
"notes": notes,
"deferred_runs": deferred_runs,
+ "rate_limited_runs": rate_limited_runs,
}
try:
previous = json.loads(reported_file.read_text())
@@ -2072,8 +2118,8 @@ def drop_reported(
return drop("unknown", unknown), drop("attention", attention),
drop("notes", notes)
-def read_deferred_runs(reported_file: Path | None) -> int | None:
- """The consecutive runs before this one that deferred job statuses; None
without a readable record."""
+def read_run_count(reported_file: Path | None, key: str) -> int | None:
+ """A count of consecutive runs the previous run recorded; None without a
readable record."""
if reported_file is None:
return None
try:
@@ -2083,10 +2129,26 @@ def read_deferred_runs(reported_file: Path | None) ->
int | None:
if not isinstance(previous, dict):
return None
# A record written before the count was kept has none.
- count = previous.get("deferred_runs", 0)
+ count = previous.get(key, 0)
return count if isinstance(count, int) and not isinstance(count, bool)
else 0
+def log_rate_limits(client: GitHubClient) -> None:
+ """Log what is left of the token's REST and GraphQL quotas; reading them
does not count against them."""
+ try:
+ resources = client.rest_get("rate_limit")["resources"]
+ # Only logged: failing to read the quotas must not fail a run that did its
work.
+ except Exception as e:
+ console_print(f"[warning]Could not read the rate limits: {e}[/]")
+ return
+ left = ", ".join(
+ f"{name} {resources[name]['remaining']}/{resources[name]['limit']}"
+ for name in ("core", "graphql")
+ if name in resources
+ )
+ console_print(f"[info]Rate limits left: {left}; rate-limited calls this
run: {client.rate_limited}[/]")
+
+
def run_reconcile(
client: GitHubClient,
config: GateConfig,
@@ -2163,7 +2225,7 @@ def run_reconcile(
f"reporting `{CI_RUN_STARTED_CONTEXT}` for project CI failed:
{type(e).__name__}"
)
job_reporter = JobStatusReporter(
- client, repo=repo, budget=max(MAX_JOB_STATUSES_PER_RUN -
ci_run_started_writes, 0)
+ client, repo=repo, budget=max(MAX_JOB_STATUSES_PER_RUN -
ci_run_started_writes, 0), now=now
)
shared_heads = find_heads_shared_across_bases(prs)
@@ -2224,10 +2286,11 @@ def run_reconcile(
unknown.append((listed.number, f"{type(e).__name__}: {e}"))
attention.extend(bypass_finder.ambiguous)
+ attention.extend(job_reporter.delayed)
notes = [bypass_finder.incomplete] if bypass_finder.incomplete else []
if ci_run_started_note:
notes.append(ci_run_started_note)
- previous_deferred_runs = read_deferred_runs(reported_file)
+ previous_deferred_runs = read_run_count(reported_file, "deferred_runs")
deferred_runs = (previous_deferred_runs or 0) + 1 if job_reporter.deferred
else 0
if job_reporter.deferred:
console_print(
@@ -2242,6 +2305,13 @@ def run_reconcile(
for number, reason in [*unknown, *attention]:
console_print(f"[warning]#{number}: {reason}[/]")
console_print(f"[info]Reconciled {len(prs)} pull requests; GitHub calls:
{dict(client.calls)}[/]")
+ log_rate_limits(client)
+ previous_rate_limited_runs = read_run_count(reported_file,
"rate_limited_runs")
+ rate_limited_runs = (previous_rate_limited_runs or 0) + 1 if
client.rate_limited else 0
+ if client.rate_limited and (
+ previous_rate_limited_runs is None or rate_limited_runs >=
RATE_LIMITED_RUNS_BEFORE_ALERT
+ ):
+ notes.append(RATE_LIMITED_NOTE)
if draft_changes_file is not None:
draft_changes_file.write_text(json.dumps([asdict(change) for change in
draft_changes]))
if reported_file is not None:
@@ -2252,6 +2322,7 @@ def run_reconcile(
notes=notes,
record=not client.dry_run,
deferred_runs=deferred_runs,
+ rate_limited_runs=rate_limited_runs,
)
has_alerts = bool(unknown or attention or notes)
if has_alerts:
diff --git a/dev/breeze/tests/test_fork_ci_github.py
b/dev/breeze/tests/test_fork_ci_github.py
index 9f6505e2954..25ae3f105c6 100644
--- a/dev/breeze/tests/test_fork_ci_github.py
+++ b/dev/breeze/tests/test_fork_ci_github.py
@@ -40,6 +40,7 @@ def test_retries_on_rate_limit(mock_run, mock_sleep):
assert client.graphql("query { ok }") == {"ok": True}
assert client.calls["graphql"] == 2
+ assert client.rate_limited == 1
mock_sleep.assert_called_once_with(30)
@@ -48,9 +49,12 @@ def test_retries_on_rate_limit(mock_run, mock_sleep):
def test_gives_up_after_the_last_attempt_without_waiting(mock_run, mock_sleep):
mock_run.return_value = _completed(1, stderr="API rate limit exceeded")
+ client = GitHubClient()
+
with pytest.raises(GitHubError, match="still rate limited after 4
attempts"):
- GitHubClient().graphql("query { ok }")
+ client.graphql("query { ok }")
assert [c.args[0] for c in mock_sleep.call_args_list] == [30, 60, 120]
+ assert client.rate_limited == 4
@pytest.mark.parametrize(
diff --git a/dev/breeze/tests/test_fork_ci_reconcile.py
b/dev/breeze/tests/test_fork_ci_reconcile.py
index eeaf790570a..9c72881dc46 100644
--- a/dev/breeze/tests/test_fork_ci_reconcile.py
+++ b/dev/breeze/tests/test_fork_ci_reconcile.py
@@ -42,6 +42,7 @@ from airflow_breeze.utils.fork_ci_messages import (
)
from airflow_breeze.utils.fork_ci_reconcile import (
JOB_STATUSES_DEFERRED_NOTE,
+ RATE_LIMITED_NOTE,
BaseGateConfigs,
BypassFinder,
DraftChange,
@@ -71,6 +72,7 @@ from airflow_breeze.utils.fork_ci_reconcile import (
is_own_pilot_change,
jobs_show_project_ci,
load_workflow_triggers,
+ log_rate_limits,
mark_older_pilot_pull_request,
parse_pull_request,
pull_request_changes_ci_files,
@@ -950,6 +952,7 @@ def client():
github = mock.create_autospec(GitHubClient, instance=True)
github.repo = REPO
github.calls = {}
+ github.rate_limited = 0
return github
@@ -1821,6 +1824,7 @@ def
test_fork_observations_use_the_fork_name_and_latest_suite(client):
},
{
"createdAt": "2026-11-02T00:00:00Z",
+ "updatedAt": "2026-11-02T00:40:00Z",
"status": "COMPLETED",
"conclusion": "SUCCESS",
"workflowRun": {
@@ -1860,7 +1864,12 @@ def
test_fork_observations_use_the_fork_name_and_latest_suite(client):
observations = fetch_fork_observations(client, [pr])
assert observations[1] == ForkObservation(
- suite_status="COMPLETED", suite_conclusion="SUCCESS", run_url="new",
run_id=77, run_attempt=2
+ suite_status="COMPLETED",
+ suite_conclusion="SUCCESS",
+ run_url="new",
+ run_id=77,
+ run_attempt=2,
+ completed_at=datetime(2026, 11, 2, 0, 40, tzinfo=UTC),
)
assert client.graphql.call_args.args[1] == {"o0": "someone", "n0":
"my-airflow", "s0": "sha1"}
@@ -2193,6 +2202,7 @@ def
test_drop_reported_reports_everything_without_a_readable_previous_run(tmp_pa
"attention": [],
"notes": [],
"deferred_runs": 0,
+ "rate_limited_runs": 0,
}
@@ -2320,7 +2330,7 @@ def _fork_job(name, status="completed",
conclusion="success", url=None):
}
-def test_build_job_statuses_reports_each_started_or_queued_job():
+def test_build_job_statuses_reports_each_finished_job():
jobs = [
_fork_job("Static checks"),
_fork_job("Postgres tests: core / A", conclusion="failure"),
@@ -2340,8 +2350,6 @@ def
test_build_job_statuses_reports_each_started_or_queued_job():
} == {
"Static checks": "success",
"Postgres tests: core / A": "failure",
- "Helm tests": "pending",
- "Kubernetes tests": "pending",
"Special tests": "failure",
}
assert statuses[f"{JOB_STATUS_CONTEXT_PREFIX}Static checks"][1] ==
"https://job/Static checks"
@@ -2420,10 +2428,11 @@ def
test_job_status_reporter_leaves_what_does_not_fit_in_the_budget_for_the_next
def test_job_status_reporter_does_not_record_a_run_in_progress(client):
- _reporter_client(client, [_fork_job("A", status="in_progress",
conclusion=None)])
+ _reporter_client(client, [_fork_job("A"), _fork_job("B",
status="in_progress", conclusion=None)])
assert JobStatusReporter(client, repo=REPO).report(_pr(), RUNNING_RUN) is
None
client.set_status.assert_called_once()
+ assert client.set_status.call_args.kwargs["context"] ==
f"{JOB_STATUS_CONTEXT_PREFIX}A"
@pytest.mark.parametrize(
@@ -2475,6 +2484,7 @@ def
test_run_reconcile_reports_job_statuses_before_classifying(
mock_classify.return_value = Plan(state=ForkState.RUNNING)
mock_reporter.return_value.report.return_value = "7:SUCCESS"
mock_reporter.return_value.deferred = []
+ mock_reporter.return_value.delayed = []
client.graphql.side_effect = [
_search_result(),
_search_result(),
@@ -2499,11 +2509,9 @@ def
test_run_reconcile_reports_job_statuses_before_classifying(
assert mock_classify.call_args.kwargs["jobs_reported"] == "7:SUCCESS"
-def test_build_job_statuses_reports_a_queued_job_like_a_running_one():
- queued = build_job_statuses([_fork_job("A", status="queued",
conclusion=None, url="u")])
- running = build_job_statuses([_fork_job("A", status="in_progress",
conclusion=None, url="u")])
-
- assert queued == running
[email protected]("status", ["queued", "in_progress", "waiting"])
+def
test_build_job_statuses_leaves_unfinished_jobs_to_the_aggregate_status(status):
+ assert build_job_statuses([_fork_job("A", status=status, conclusion=None,
url="u")]) == {}
def _stale(state):
@@ -2586,6 +2594,7 @@ def
test_run_reconcile_classifies_when_reporting_job_statuses_fails(
mock_classify.return_value = Plan(state=ForkState.RUNNING)
mock_reporter.return_value.report.side_effect = GitHubError("HTTP 502")
mock_reporter.return_value.deferred = []
+ mock_reporter.return_value.delayed = []
assert _reconcile_with_reporter(client, tmp_path) is True
@@ -2598,6 +2607,7 @@ def
test_run_reconcile_classifies_when_reporting_job_statuses_fails(
def test_run_reconcile_without_job_statuses(mock_classify, mock_reporter,
client, tmp_path):
mock_classify.return_value = Plan(state=ForkState.RUNNING)
mock_reporter.return_value.deferred = []
+ mock_reporter.return_value.delayed = []
_reconcile_with_reporter(client, tmp_path,
config=dataclasses.replace(CONFIG, job_statuses=False))
@@ -2622,6 +2632,7 @@ def
test_run_reconcile_reports_deferred_job_statuses_to_slack(
mock_classify.return_value = Plan(state=ForkState.RUNNING)
mock_reporter.return_value.report.return_value = None
mock_reporter.return_value.deferred = deferred
+ mock_reporter.return_value.delayed = []
assert _reconcile_with_reporter(client, tmp_path) is has_alerts
@@ -2642,6 +2653,7 @@ def
test_run_reconcile_reports_deferred_job_statuses_only_when_deferring_persist
alerts = []
for deferred in ([1], [1], [], [1], [1], [1]):
mock_reporter.return_value.deferred = deferred
+ mock_reporter.return_value.delayed = []
alerts.append(_reconcile_with_reporter(client, tmp_path,
reported_file=reported_file))
assert alerts == [False, False, False, False, False, True]
@@ -3155,7 +3167,7 @@ def
test_run_reconcile_retries_a_pull_request_whose_head_moved(client, tmp_path)
@pytest.mark.parametrize(
("fails", "job_status_budget"),
- [pytest.param(False, 58, id="reported"), pytest.param(True, 30,
id="fails")],
+ [pytest.param(False, 38, id="reported"), pytest.param(True, 20,
id="fails")],
)
@mock.patch("airflow_breeze.utils.fork_ci_reconcile.JobStatusReporter",
autospec=True)
@mock.patch("airflow_breeze.utils.fork_ci_reconcile.report_project_ci_runs_started",
autospec=True)
@@ -3169,6 +3181,7 @@ def
test_run_reconcile_marks_older_pilot_pull_requests_and_reports_project_ci_ru
mock_report.return_value = [7, 8]
mock_reporter.return_value.report.return_value = None
mock_reporter.return_value.deferred = []
+ mock_reporter.return_value.delayed = []
client.graphql.side_effect = [
_search_result(),
_search_result(),
@@ -3191,7 +3204,7 @@ def
test_run_reconcile_marks_older_pilot_pull_requests_and_reports_project_ci_ru
assert mock_classify.call_args.args[0].ci_run_at == datetime(2026, 11, 2,
tzinfo=UTC)
mock_report.assert_called_once_with(client, repo=REPO, skip={1})
- mock_reporter.assert_called_once_with(client, repo=REPO,
budget=job_status_budget)
+ mock_reporter.assert_called_once_with(client, repo=REPO,
budget=job_status_budget, now=NOW)
assert has_alerts is fails
if fails:
assert "reporting `CI run started` for project CI failed: GitHubError"
in output.read_text()
@@ -3232,3 +3245,99 @@ def
test_older_pilot_pull_request_stays_checked_for_tampering_after_its_gated_ru
)
assert find_bypass.called
+
+
[email protected](
+ ("resources", "logged"),
+ [
+ pytest.param(
+ {"core": {"remaining": 900, "limit": 1000}, "graphql":
{"remaining": 990, "limit": 1000}},
+ "core 900/1000, graphql 990/1000",
+ id="read",
+ ),
+ pytest.param(
+ GitHubError("gh api failed: HTTP 500"), "Could not read the rate
limits", id="unreadable"
+ ),
+ ],
+)
[email protected]("airflow_breeze.utils.fork_ci_reconcile.console_print",
autospec=True)
+def test_log_rate_limits(mock_print, client, resources, logged):
+ client.rest_get.side_effect = (
+ [resources] if isinstance(resources, Exception) else [{"resources":
resources}]
+ )
+
+ log_rate_limits(client)
+
+ client.rest_get.assert_called_once_with("rate_limit")
+ assert logged in mock_print.call_args.args[0]
+
+
[email protected]("airflow_breeze.utils.fork_ci_reconcile.JobStatusReporter",
autospec=True)
[email protected]("airflow_breeze.utils.fork_ci_reconcile.classify", autospec=True)
+def test_run_reconcile_reports_rate_limiting_only_when_it_persists(
+ mock_classify, mock_reporter, client, tmp_path
+):
+ mock_classify.return_value = Plan(state=ForkState.RUNNING)
+ mock_reporter.return_value.report.return_value = None
+ mock_reporter.return_value.deferred = []
+ mock_reporter.return_value.delayed = []
+ client.dry_run = False
+ reported_file = tmp_path / "reported.json"
+ reported_file.write_text("{}")
+ alerts = []
+ for rate_limited in (1, 0, 3, 2):
+ client.rate_limited = rate_limited
+ alerts.append(_reconcile_with_reporter(client, tmp_path,
reported_file=reported_file))
+
+ assert alerts == [False, False, False, True]
+ assert RATE_LIMITED_NOTE in (tmp_path / "s.json").read_text()
+ assert json.loads(reported_file.read_text())["rate_limited_runs"] == 2
+
+
[email protected](
+ ("completed_minutes_ago", "delayed"),
+ [
+ pytest.param(61, True, id="finished-over-an-hour-ago"),
+ pytest.param(30, False, id="finished-recently"),
+ pytest.param(None, False, id="still-running"),
+ ],
+)
+def
test_job_status_reporter_reports_statuses_deferred_long_after_the_run_finished(
+ client, completed_minutes_ago, delayed
+):
+ fork = dataclasses.replace(
+ COMPLETED_RUN,
+ completed_at=None
+ if completed_minutes_ago is None
+ else NOW - timedelta(minutes=completed_minutes_ago),
+ )
+ reporter = JobStatusReporter(client, repo=REPO, budget=0, now=NOW)
+
+ assert reporter.report(_pr(), fork) is None
+
+ assert reporter.deferred == [42]
+ assert reporter.delayed == (
+ [
+ (
+ 42,
+ f"job statuses of fork run {fork.run_url} are still not all
posted more than 1:00:00 after it finished",
+ )
+ ]
+ if delayed
+ else []
+ )
+
+
[email protected]("airflow_breeze.utils.fork_ci_reconcile.JobStatusReporter",
autospec=True)
[email protected]("airflow_breeze.utils.fork_ci_reconcile.classify", autospec=True)
+def test_run_reconcile_reports_delayed_job_statuses_as_attention_items(
+ mock_classify, mock_reporter, client, tmp_path
+):
+ mock_classify.return_value = Plan(state=ForkState.RUNNING)
+ mock_reporter.return_value.report.return_value = None
+ mock_reporter.return_value.deferred = [1]
+ mock_reporter.return_value.delayed = [(1, "job statuses of fork run R are
still not all posted")]
+
+ assert _reconcile_with_reporter(client, tmp_path) is True
+
+ assert "job statuses of fork run R are still not all posted" in (tmp_path
/ "s.json").read_text()
