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 05d86b21c28 Post the first fork CI comment without waiting for the
next run (#76510)
05d86b21c28 is described below
commit 05d86b21c282d9b18bc91825d0875a5276b4fbee
Author: Andrew Chang <[email protected]>
AuthorDate: Sat Nov 7 08:05:50 2026 +0800
Post the first fork CI comment without waiting for the next run (#76510)
* Comment on newly gated pull requests in the run that labels them
The fork CI reconciler labels newly gated pull requests, then searches
for the pull requests that carry the label. The search index often does
not list a pull request labelled seconds before. Its state comment and
setup instructions then waited for the next run, about 15 minutes later.
6 of the 19 pilot pull requests opened since 2026-10-31 waited for this
extra run.
* Leave pull requests for the next run when the repository read is null
A partial GraphQL response with a null repository made
fetch_open_pull_requests raise AttributeError, which aborted the whole
reconciler run instead of leaving the pull requests for the next run.
Generated-by: Claude Opus 5
---------
Co-authored-by: Jarek Potiuk <[email protected]>
---
.../doc/ci/08_fork_ci_for_external_contributors.md | 11 ++-
.../src/airflow_breeze/utils/fork_ci_reconcile.py | 77 ++++++++++++---
dev/breeze/tests/test_fork_ci_reconcile.py | 104 +++++++++++++++++++++
3 files changed, 176 insertions(+), 16 deletions(-)
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 f1007448d61..e60311dfa06 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
@@ -501,8 +501,12 @@ state targetUrl description } } } } }`), and the
reconciler's own state comment
(`comments(first: 100)`, the earliest comment by `github-actions` with the
marker, paging
through the remaining comments of the rare pull request with more). The
reconciler creates that
comment on first contact and only edits it afterwards, so a marker posted by
anyone else, or
-echoed later by another workflow, is ignored. The fork state is then fetched
in batches of 50 with one GraphQL
-query per batch using aliases. Owner and name come from the pull request's
+echoed later by another workflow, is ignored. The search index can take
minutes to list a pull
+request under the label pass 1 has just added, so the pull requests pass 1
labelled that the
+search does not list yet are read by number with the same fields, 25 per
query, and reconciled in
+the same run. If that read fails, they are left for the next run. The fork
state is then
+fetched in batches of 50 with one GraphQL query per batch using aliases. Owner
and name come
+from the pull request's
`headRepository { nameWithOwner }`, never a fixed `airflow`, because a fork
can be renamed or
created as `airflow-1` on a name collision:
@@ -1064,7 +1068,8 @@ files, a jobs call and a compare call only per long
`ci-amd.yml` run of a gated
one changed-files call per head commit that turns green, plus one mutation per
actual state
change. Job statuses add a jobs call and a status call per pull request with a
fork run in
progress or a completed run not yet reported, and up to two status mutations
per job, at most
-`MAX_JOB_STATUSES_PER_RUN` per run; see "Write budget". Well inside
+`MAX_JOB_STATUSES_PER_RUN` per run; see "Write budget". Pull requests that
pass 1 labelled and
+the search does not list yet add one GraphQL call per 25. Well inside
the `GITHUB_TOKEN` limits at a 15-minute cadence. The reconciler logs its call
counts so the
budget stays observable.
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 c9a6865cd96..4e2f8cfe6ef 100644
--- a/dev/breeze/src/airflow_breeze/utils/fork_ci_reconcile.py
+++ b/dev/breeze/src/airflow_breeze/utils/fork_ci_reconcile.py
@@ -832,11 +832,8 @@ query($q: String!, $cursor: String) {
}
"""
-PASS_TWO_QUERY = """
-query($q: String!, $cursor: String) {
- search(type: ISSUE, first: %(page)d, after: $cursor, query: $q) {
- pageInfo { hasNextPage endCursor }
- nodes { ... on PullRequest {
+PASS_TWO_FIELDS = """
+fragment PassTwoFields on PullRequest {
number id isDraft createdAt authorAssociation baseRefName headRefName
headRefOid
author { login }
headRepository { nameWithOwner }
@@ -865,10 +862,35 @@ query($q: String!, $cursor: String) {
pageInfo { hasNextPage endCursor }
nodes { databaseId body author { login } }
}
- } }
+}
+"""
+
+PASS_TWO_QUERY = (
+ """
+query($q: String!, $cursor: String) {
+ search(type: ISSUE, first: %(page)d, after: $cursor, query: $q) {
+ pageInfo { hasNextPage endCursor }
+ nodes { ...PassTwoFields }
}
}
"""
+ + PASS_TWO_FIELDS
+)
+
+PULL_REQUESTS_BY_NUMBER_QUERY = (
+ """
+query($owner: String!, $name: String!) {
+ repository(owner: $owner, name: $name) { %(pull_requests)s }
+}
+"""
+ + PASS_TWO_FIELDS
+)
+
+PASS_TWO_CONTEXTS = {
+ "context": STATUS_CONTEXT,
+ "ci_passed": CI_PASSED_CONTEXT,
+ "ci_run_started": CI_RUN_STARTED_CONTEXT,
+}
def _labels(node: dict[str, Any]) -> tuple[str, ...]:
@@ -1112,6 +1134,35 @@ def _search(client: GitHubClient, query: str,
search_query: str) -> Iterable[dic
cursor = result["pageInfo"]["endCursor"]
+def fetch_open_pull_requests(
+ client: GitHubClient, numbers: Sequence[int], *, repo: str
+) -> list[dict[str, Any]]:
+ """
+ Return the pass 2 nodes of the pull requests among these that are still
open.
+
+ Reads them by number rather than through the search index, which can take
minutes to list a
+ pull request under a label added moments before.
+ """
+ owner, name = repo.split("/", 1)
+ nodes: list[dict[str, Any]] = []
+ for start in range(0, len(numbers), PASS_TWO_PAGE_SIZE):
+ batch = numbers[start : start + PASS_TWO_PAGE_SIZE]
+ pull_requests = " ".join(
+ f"pr{index}: pullRequest(number: {number}) {{ state
...PassTwoFields }}"
+ for index, number in enumerate(batch)
+ )
+ result = (
+ client.graphql(
+ PULL_REQUESTS_BY_NUMBER_QUERY % {"pull_requests":
pull_requests, **PASS_TWO_CONTEXTS},
+ {"owner": owner, "name": name},
+ allow_partial=True,
+ )["repository"]
+ or {}
+ )
+ nodes.extend(node for node in result.values() if node and
node["state"] == "OPEN")
+ return nodes
+
+
def _is_not_found(error: GitHubError) -> bool:
return "404" in str(error) or "Not Found" in str(error)
@@ -2235,16 +2286,16 @@ def run_reconcile(
parse_pull_request(node)
for node in _search(
client,
- PASS_TWO_QUERY
- % {
- "page": PASS_TWO_PAGE_SIZE,
- "context": STATUS_CONTEXT,
- "ci_passed": CI_PASSED_CONTEXT,
- "ci_run_started": CI_RUN_STARTED_CONTEXT,
- },
+ PASS_TWO_QUERY % {"page": PASS_TWO_PAGE_SIZE, **PASS_TWO_CONTEXTS},
f'repo:{repo} is:pr is:open label:"{LABEL_FORK_CI_REQUIRED}"',
)
]
+ found = {pr.number for pr in prs}
+ unlisted = [number for number in labelled if number not in found]
+ try:
+ prs.extend(parse_pull_request(node) for node in
fetch_open_pull_requests(client, unlisted, repo=repo))
+ except GitHubError as e:
+ console_print(f"[warning]Pull requests {unlisted} are left for the
next run: {e}[/]")
observations = fetch_fork_observations(client, prs)
def is_gated_base(base_ref: str) -> bool:
diff --git a/dev/breeze/tests/test_fork_ci_reconcile.py
b/dev/breeze/tests/test_fork_ci_reconcile.py
index 81962145a43..de08f6a6f46 100644
--- a/dev/breeze/tests/test_fork_ci_reconcile.py
+++ b/dev/breeze/tests/test_fork_ci_reconcile.py
@@ -63,6 +63,7 @@ from airflow_breeze.utils.fork_ci_reconcile import (
discover,
drop_reported,
fetch_fork_observations,
+ fetch_open_pull_requests,
fetch_workflow_state,
find_ci_run_since,
find_heads_shared_across_bases,
@@ -3240,6 +3241,109 @@ def
test_gate_change_finder_aborts_when_the_pull_request_changed(client, head_sh
finder.find(_pr())
+NEWLY_GATED = {**_pr_node(1), "labels": {"nodes": []}}
+
+
+def _graphql_after_gating_pull_request_1(*, listed: list[dict], by_number:
dict | Exception | None):
+ def graphql(query, variables=None, *, allow_partial=False):
+ if "pullRequest(number:" in query:
+ if by_number is None:
+ raise AssertionError("read by number although the search
listed it")
+ if isinstance(by_number, Exception):
+ raise by_number
+ return by_number
+ if "q" not in variables:
+ return {f"pr{key[1:]}": {"object": None} for key in variables if
key.startswith("s")}
+ if "updated:>=" in variables["q"]:
+ return _search_result(NEWLY_GATED)
+ if variables["q"] == f'repo:{REPO} is:pr is:open label:"fork ci
required"':
+ return _search_result(*listed)
+ return _search_result()
+
+ return graphql
+
+
+def _rest_get_for_pull_request_1(path, paginate=False, items_key=None):
+ if path.endswith("/actions/workflows/ci-amd.yml"):
+ return {"state": "active"}
+ if path == f"repos/{REPO}/pulls/1/files?per_page=100":
+ return []
+ if path == f"repos/{REPO}/pulls/1":
+ return {"head": {"sha": "sha1"}, "base": {"ref": "main"}}
+ raise AssertionError(path)
+
+
[email protected](
+ ("listed", "by_number", "commented"),
+ [
+ pytest.param(
+ [], {"repository": {"pr0": {**_pr_node(1), "state": "OPEN"}}},
[1], id="not-yet-listed-by-search"
+ ),
+ pytest.param([_pr_node(1)], None, [1], id="listed-by-search"),
+ pytest.param(
+ [], {"repository": {"pr0": {**_pr_node(1), "state": "CLOSED"}}},
[], id="closed-since-labelled"
+ ),
+ pytest.param([], {"repository": {"pr0": None}}, [],
id="deleted-since-labelled"),
+ pytest.param([], {"repository": None}, [],
id="repository-not-resolved"),
+ ],
+)
+def test_run_reconcile_reconciles_pull_requests_labelled_in_the_same_run(
+ client, tmp_path, listed, by_number, commented
+):
+ client.graphql.side_effect =
_graphql_after_gating_pull_request_1(listed=listed, by_number=by_number)
+ client.rest_get_pages.return_value = [{"total_count": 0, "workflow_runs":
[]}]
+ client.rest_get.side_effect = _rest_get_for_pull_request_1
+
+ has_alerts = run_reconcile(
+ client,
+ CONFIG,
+ now=NOW,
+ run_url="https://run";,
+ output_file=tmp_path / "s.json",
+ workflows_dir=tmp_path,
+ )
+
+ assert has_alerts is False
+ client.add_labels.assert_called_once_with(1, ["fork ci required"])
+ assert [call.args[0] for call in client.create_comment.call_args_list] ==
commented
+
+
+def
test_run_reconcile_leaves_pull_requests_it_cannot_read_for_the_next_run(client,
tmp_path):
+ client.graphql.side_effect = _graphql_after_gating_pull_request_1(
+ listed=[], by_number=GitHubError("gh api failed: HTTP 502")
+ )
+ client.rest_get_pages.return_value = [{"total_count": 0, "workflow_runs":
[]}]
+ client.rest_get.side_effect = _rest_get_for_pull_request_1
+
+ has_alerts = run_reconcile(
+ client,
+ CONFIG,
+ now=NOW,
+ run_url="https://run";,
+ output_file=tmp_path / "s.json",
+ workflows_dir=tmp_path,
+ )
+
+ assert has_alerts is False
+ client.create_comment.assert_not_called()
+
+
[email protected]("airflow_breeze.utils.fork_ci_reconcile.PASS_TWO_PAGE_SIZE", 2)
+def test_fetch_open_pull_requests_reads_in_batches(client):
+ client.graphql.side_effect = [
+ {"repository": {"pr0": {"number": 1, "state": "OPEN"}, "pr1":
{"number": 2, "state": "OPEN"}}},
+ {"repository": {"pr0": {"number": 3, "state": "OPEN"}}},
+ ]
+
+ nodes = fetch_open_pull_requests(client, [1, 2, 3], repo=REPO)
+
+ assert [node["number"] for node in nodes] == [1, 2, 3]
+ queries = [call.args[0] for call in client.graphql.call_args_list]
+ assert "pr1: pullRequest(number: 2)" in queries[0]
+ assert "pr0: pullRequest(number: 3)" in queries[1]
+ assert "number: 3" not in queries[0]
+
+
def test_run_reconcile_retries_a_pull_request_whose_head_moved(client,
tmp_path):
output = tmp_path / "slack-message.json"
client.graphql.side_effect = [
