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 e5f2c87a3ab Cancel the workflow runs of pull requests that have
already merged (#75619)
e5f2c87a3ab is described below
commit e5f2c87a3ab06c7dde2b84c64d9fd9a555b3ad12
Author: Jarek Potiuk <[email protected]>
AuthorDate: Sun Oct 25 20:10:11 2026 +0100
Cancel the workflow runs of pull requests that have already merged (#75619)
Merging a pull request does not stop its pull_request runs, so they keep
using the project's GitHub-hosted minutes and self-hosted runners for a
result nobody reads once the change is on its target branch. Until now
maintainers cancelled such runs by hand, when they noticed them.
Generated-by: Claude Opus 5
---
.github/fork-ci-gate.toml | 5 +
.github/workflows/fork-ci-monitor.yml | 4 +-
dev/breeze/doc/08_ci_tasks.rst | 3 +-
.../doc/ci/08_fork_ci_for_external_contributors.md | 33 ++-
.../src/airflow_breeze/utils/fork_ci_gate.py | 5 +
.../src/airflow_breeze/utils/fork_ci_github.py | 8 +
.../src/airflow_breeze/utils/fork_ci_reconcile.py | 8 +
.../utils/merged_pull_request_runs.py | 211 ++++++++++++++++
dev/breeze/tests/test_fork_ci_gate.py | 8 +
dev/breeze/tests/test_fork_ci_github.py | 17 ++
dev/breeze/tests/test_fork_ci_reconcile.py | 40 ++++
dev/breeze/tests/test_merged_pull_request_runs.py | 266 +++++++++++++++++++++
12 files changed, 604 insertions(+), 4 deletions(-)
diff --git a/.github/fork-ci-gate.toml b/.github/fork-ci-gate.toml
index 6fc7a0de1ce..6119f610ef2 100644
--- a/.github/fork-ci-gate.toml
+++ b/.github/fork-ci-gate.toml
@@ -34,6 +34,11 @@ close_after_days = 7
# status is posted either way.
job_statuses = true
+# Cancel the unfinished `pull_request` runs of pull requests that have already
merged, of any
+# author, so they stop using GitHub-hosted minutes and self-hosted runners for
a result nobody
+# reads. The fork CI monitor does it on its schedule.
+cancel_merged_runs = true
+
# Organisations exempted under AIP-120 ยง2.2. Keys are display names used in
comments,
# values are GitHub logins. Changes go through a reviewed pull request.
[allowlist]
diff --git a/.github/workflows/fork-ci-monitor.yml
b/.github/workflows/fork-ci-monitor.yml
index ec34b9615b0..278787be5af 100644
--- a/.github/workflows/fork-ci-monitor.yml
+++ b/.github/workflows/fork-ci-monitor.yml
@@ -41,8 +41,8 @@ jobs:
name: "Reconcile pull requests gated to fork CI"
if: github.repository == 'apache/airflow'
permissions:
- # Listing the runs and jobs of the gate tampering check, and deleting
the runs that only
- # gated a pull request to fork CI.
+ # Listing the runs and jobs of the gate tampering check, deleting the
runs that only gated a
+ # pull request to fork CI, and cancelling the unfinished runs of merged
pull requests.
actions: write
contents: read
# Labels, comments and closing of gated pull requests.
diff --git a/dev/breeze/doc/08_ci_tasks.rst b/dev/breeze/doc/08_ci_tasks.rst
index ca16b4483fe..b1bf0d9f5fb 100644
--- a/dev/breeze/doc/08_ci_tasks.rst
+++ b/dev/breeze/doc/08_ci_tasks.rst
@@ -188,7 +188,8 @@ Pull requests from external contributors can run their CI
in the contributor's f
gated pull requests ``fork ci required``, keeps them in draft until the fork
run is green, maintains
one state comment and a ``Fork CI / Tests (AMD)`` commit status linking to the
run in the fork, closes
pull requests without fork CI activity for a week, closes gated pull requests
that made project CI run
-by changing workflows, and writes everything it could not classify to a Slack
message. With
+by changing workflows, and writes everything it could not classify to a Slack
message. It also cancels
+the unfinished workflow runs of pull requests, of any author, that have
already merged. With
``--dry-run`` it prints the changes instead of applying them. See
`Fork CI for external contributors
<ci/08_fork_ci_for_external_contributors.md>`_ for the design.
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 799a233e9ca..1f398373944 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
@@ -787,6 +787,36 @@ name within its time limit, and its run is the evidence. A
re-run is kept too: d
deletes every attempt, and an earlier attempt may have run project CI. Nothing
is deleted for a
pull request in the `gate_bypassed` state. A dry run only reports the
deletions.
+### Cancelling runs of merged pull requests
+
+Merging a pull request does not stop its `pull_request` runs. They keep using
`apache/airflow`'s
+capacity, GitHub-hosted minutes from the ASF's allocation and the self-hosted
runners, for a
+result nobody reads once the change is on its target branch. This is not
specific to fork CI; the
+monitor does it because it already runs every 15 minutes with `actions: write`.
+
+Each monitor run lists the `pull_request` runs of `apache/airflow` that are
`queued`,
+`in_progress`, `waiting`, `pending` or `requested`, and looks up the pull
requests of their head
+branches in one GraphQL query per 25 branches, paging through every pull
request of a branch: a
+common branch name such as `main` has pull requests from many forks, and an
open one with the
+run's head must not be missed. A run is cancelled when:
+
+- a merged pull request has the run's head repository, branch and commit (runs
of pull requests
+ from forks do not list their pull request, so they are matched this way);
+- no open pull request has the same head, since a branch can be reused for a
new pull request;
+- the run was created before the merge. A workflow triggered by the merge
itself, such as one
+ on the `closed` event, is meant to run.
+
+It applies to the pull requests of every author, not only gated ones. A run is
cancelled with
+`POST /repos/apache/airflow/actions/runs/<id>/cancel`; one still `in_progress`
30 minutes after the
+merge, for example because a step ignores the cancellation, is force-cancelled
+(`.../force-cancel`). A run that finished in the meantime (HTTP 409) is not a
problem; any other
+failure is reported to Slack with the pull request, and a failure of the whole
pass is reported
+as a note, without affecting the reconciliation of gated pull requests. A dry
run only reports
+the cancellations. `cancel_merged_runs = false` in `.github/fork-ci-gate.toml`
turns it off.
+
+Only runs of the merged head commit are matched. A run of an earlier commit of
the pull request is
+normally cancelled by the workflow's concurrency group when the next commit is
pushed.
+
### Far behind the base branch
A fork run tests the pushed commit as it is, while project CI tests it merged
with the base
@@ -841,7 +871,8 @@ Per run, with ~700 gated pull requests: ~28 GraphQL calls
for pass 2, ~14 for fo
calls for the recent `pull_request` runs, one REST call per base branch for
its workflow
files, a jobs call and a compare call only per long `ci-amd.yml` run of a
gated pull request,
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
+change. Cancelling runs of merged pull requests adds five paginated run
listings, one GraphQL
+query per 25 head branches with unfinished runs, and one mutation per
cancelled run. 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
the `GITHUB_TOKEN` limits at a 15-minute cadence. The reconciler logs its call
counts so the
diff --git a/dev/breeze/src/airflow_breeze/utils/fork_ci_gate.py
b/dev/breeze/src/airflow_breeze/utils/fork_ci_gate.py
index 8ce90a34d97..dc05a3b67a5 100644
--- a/dev/breeze/src/airflow_breeze/utils/fork_ci_gate.py
+++ b/dev/breeze/src/airflow_breeze/utils/fork_ci_gate.py
@@ -129,6 +129,7 @@ class GateConfig:
allowlist: Mapping[str, tuple[str, ...]] = field(default_factory=dict)
pilot: Mapping[str, datetime] = field(default_factory=dict)
job_statuses: bool = True
+ cancel_merged_runs: bool = True
def is_allowlisted(self, login: str) -> bool:
wanted = login.casefold()
@@ -177,6 +178,9 @@ def parse_config(text: str) -> GateConfig:
job_statuses = data.get("job_statuses", True)
if not isinstance(job_statuses, bool):
raise ValueError(f"job_statuses must be true or false, got
{job_statuses!r}")
+ cancel_merged_runs = data.get("cancel_merged_runs", True)
+ if not isinstance(cancel_merged_runs, bool):
+ raise ValueError(f"cancel_merged_runs must be true or false, got
{cancel_merged_runs!r}")
return GateConfig(
enabled=enabled,
enabled_since=enabled_since,
@@ -184,6 +188,7 @@ def parse_config(text: str) -> GateConfig:
allowlist=allowlist,
pilot=pilot,
job_statuses=job_statuses,
+ cancel_merged_runs=cancel_merged_runs,
)
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 73a44503acb..fe3fdfb651e 100644
--- a/dev/breeze/src/airflow_breeze/utils/fork_ci_github.py
+++ b/dev/breeze/src/airflow_breeze/utils/fork_ci_github.py
@@ -172,6 +172,14 @@ class GitHubClient:
None,
)
+ def cancel_run(self, pr: int, run_id: int, *, force: bool = False) -> None:
+ action = "force-cancel" if force else "cancel"
+ self._mutate(
+ f"#{pr}: {action} workflow run {run_id}",
+ ["api", "-X", "POST",
f"repos/{self.repo}/actions/runs/{run_id}/{action}"],
+ None,
+ )
+
def set_status(
self,
pr: int,
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 6b54f6c7fac..00698b78494 100644
--- a/dev/breeze/src/airflow_breeze/utils/fork_ci_reconcile.py
+++ b/dev/breeze/src/airflow_breeze/utils/fork_ci_reconcile.py
@@ -74,6 +74,7 @@ from airflow_breeze.utils.fork_ci_messages import (
tamper_close_comment,
ungated_comment,
)
+from airflow_breeze.utils.merged_pull_request_runs import
cancel_merged_pull_request_runs
CI_WORKFLOW_FILE = "ci-amd.yml"
# The files the gate job of ci-amd.yml reads from a pull request's base branch.
@@ -1547,6 +1548,13 @@ def run_reconcile(
unknown.append((listed.number, f"{type(e).__name__}: {e}"))
notes = [bypass_finder.incomplete] if bypass_finder.incomplete else []
+ if config.cancel_merged_runs:
+ # Not tied to gated pull requests; a failure here must not hide the
reconciliation's own.
+ try:
+ unknown.extend(cancel_merged_pull_request_runs(client, repo=repo,
now=now))
+ except Exception as e:
+ console_print(f"[error]Cancelling runs of merged pull requests
failed: {e}[/]")
+ notes.append(f"cancelling runs of merged pull requests failed:
{type(e).__name__}: {e}")
if job_reporter.deferred:
console_print(
f"[info]Job statuses of {len(job_reporter.deferred)} pull requests
left for the next run: "
diff --git a/dev/breeze/src/airflow_breeze/utils/merged_pull_request_runs.py
b/dev/breeze/src/airflow_breeze/utils/merged_pull_request_runs.py
new file mode 100644
index 00000000000..7eed315630b
--- /dev/null
+++ b/dev/breeze/src/airflow_breeze/utils/merged_pull_request_runs.py
@@ -0,0 +1,211 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+"""
+Cancels the workflow runs that are still going for pull requests that have
already merged.
+
+Merging a pull request does not stop its `pull_request` runs, which keep using
the project's
+GitHub-hosted minutes and self-hosted runners for a result nobody reads. The
fork CI monitor calls
+this on its schedule. See "Cancelling runs of merged pull requests" in
+dev/breeze/doc/ci/08_fork_ci_for_external_contributors.md.
+"""
+
+from __future__ import annotations
+
+from collections.abc import Iterable
+from dataclasses import dataclass
+from datetime import datetime, timedelta
+from typing import TYPE_CHECKING, Any
+
+from airflow_breeze.utils.console import console_print
+from airflow_breeze.utils.fork_ci_gate import parse_github_datetime
+
+if TYPE_CHECKING:
+ from airflow_breeze.utils.fork_ci_github import GitHubClient
+
+# The statuses of a run that has not finished. A run waiting for approval is
"completed".
+UNFINISHED_RUN_STATUSES = ("queued", "in_progress", "waiting", "pending",
"requested")
+# A cancelled run whose jobs keep running, for example in a step that ignores
the signal, is
+# force-cancelled once this long has passed since the merge, which leaves a
normal cancel at least
+# one monitor run to work.
+FORCE_CANCEL_AFTER = timedelta(minutes=30)
+BRANCH_BATCH_SIZE = 25
+
+
+@dataclass(frozen=True)
+class UnfinishedRun:
+ id: int
+ workflow: str
+ status: str
+ head_repo: str
+ head_branch: str
+ head_sha: str
+ created_at: datetime
+
+
+@dataclass(frozen=True)
+class PullRequestHead:
+ number: int
+ state: str
+ head_repo: str
+ head_sha: str
+ merged_at: datetime | None
+
+
+def list_unfinished_runs(client: GitHubClient, repo: str) ->
list[UnfinishedRun]:
+ runs = []
+ for status in UNFINISHED_RUN_STATUSES:
+ for page in client.rest_get_pages(
+
f"repos/{repo}/actions/runs?event=pull_request&status={status}&per_page=100"
+ ):
+ for run in page["workflow_runs"]:
+ runs.append(
+ UnfinishedRun(
+ id=run["id"],
+ workflow=run.get("name") or "",
+ status=run["status"],
+ head_repo=(run.get("head_repository") or
{}).get("full_name") or "",
+ head_branch=run.get("head_branch") or "",
+ head_sha=run["head_sha"],
+ created_at=parse_github_datetime(run["created_at"]),
# type: ignore[arg-type]
+ )
+ )
+ return runs
+
+
+PULL_REQUEST_FIELDS = (
+ "pageInfo { hasNextPage endCursor } "
+ "nodes { number state mergedAt headRefOid headRepository { nameWithOwner }
}"
+)
+BRANCH_PAGE_QUERY = (
+ "query($owner: String!, $name: String!, $branch: String!, $cursor: String)
"
+ "{ repository(owner: $owner, name: $name) { pullRequests(headRefName:
$branch, first: 100, "
+ f"after: $cursor) {{ {PULL_REQUEST_FIELDS} }} }} }}"
+)
+
+
+def _parse_pull_requests(connection: dict[str, Any] | None) ->
list[PullRequestHead]:
+ return [
+ PullRequestHead(
+ number=node["number"],
+ state=node["state"],
+ head_repo=(node.get("headRepository") or {}).get("nameWithOwner")
or "",
+ head_sha=node["headRefOid"],
+ merged_at=parse_github_datetime(node.get("mergedAt") or ""),
+ )
+ for node in (connection or {}).get("nodes") or []
+ ]
+
+
+def find_pull_requests_by_branch(
+ client: GitHubClient, repo: str, branches: Iterable[str]
+) -> dict[str, list[PullRequestHead]]:
+ """
+ Every pull request of each head branch name, in any state and from any
head repository.
+
+ All of them are read, not only the latest: a common branch name such as
``main`` has pull
+ requests from many forks, and an open one with a run's head must not be
missed.
+ """
+ owner, name = repo.split("/", 1)
+ found: dict[str, list[PullRequestHead]] = {}
+ unique = sorted(set(branches))
+ for start in range(0, len(unique), BRANCH_BATCH_SIZE):
+ batch = unique[start : start + BRANCH_BATCH_SIZE]
+ declarations = ", ".join(f"$b{index}: String!" for index in
range(len(batch)))
+ fields = " ".join(
+ f"b{index}: pullRequests(headRefName: $b{index}, first: 100) {{
{PULL_REQUEST_FIELDS} }}"
+ for index in range(len(batch))
+ )
+ data = client.graphql(
+ f"query($owner: String!, $name: String!, {declarations}) "
+ f"{{ repository(owner: $owner, name: $name) {{ {fields} }} }}",
+ {"owner": owner, "name": name, **{f"b{index}": branch for index,
branch in enumerate(batch)}},
+ )["repository"]
+ for index, branch in enumerate(batch):
+ connection = data[f"b{index}"] or {}
+ found[branch] = _parse_pull_requests(connection)
+ while (connection.get("pageInfo") or {}).get("hasNextPage"):
+ connection = client.graphql(
+ BRANCH_PAGE_QUERY,
+ {
+ "owner": owner,
+ "name": name,
+ "branch": branch,
+ "cursor": connection["pageInfo"]["endCursor"],
+ },
+ )["repository"]["pullRequests"]
+ found[branch].extend(_parse_pull_requests(connection))
+ return found
+
+
+def find_merged_pull_request(
+ run: UnfinishedRun, pull_requests: Iterable[PullRequestHead]
+) -> PullRequestHead | None:
+ """
+ The merged pull request a run was started for, if no open pull request has
the same head.
+
+ A run is matched by head repository, branch and commit, because runs of
pull requests from
+ forks do not list their pull request. Only runs started before the merge
count: a workflow
+ triggered by the merge itself, such as one on `closed`, is meant to run.
+ """
+ same_head = [
+ pr
+ for pr in pull_requests
+ if pr.head_repo.casefold() == run.head_repo.casefold() and pr.head_sha
== run.head_sha
+ ]
+ if any(pr.state == "OPEN" for pr in same_head):
+ return None
+ return next(
+ (
+ pr
+ for pr in same_head
+ if pr.state == "MERGED" and pr.merged_at is not None and
run.created_at < pr.merged_at
+ ),
+ None,
+ )
+
+
+def cancel_merged_pull_request_runs(
+ client: GitHubClient, *, repo: str, now: datetime
+) -> list[tuple[int, str]]:
+ """Cancel the unfinished runs of merged pull requests; return ``(pull
request, problem)`` for Slack."""
+ runs = list_unfinished_runs(client, repo)
+ if not runs:
+ return []
+ pull_requests = find_pull_requests_by_branch(client, repo,
(run.head_branch for run in runs))
+ problems: list[tuple[int, str]] = []
+ cancelled = 0
+ for run in runs:
+ merged = find_merged_pull_request(run,
pull_requests.get(run.head_branch, []))
+ if merged is None:
+ continue
+ force = (
+ run.status == "in_progress"
+ and merged.merged_at is not None
+ and now - merged.merged_at >= FORCE_CANCEL_AFTER
+ )
+ try:
+ client.cancel_run(merged.number, run.id, force=force)
+ cancelled += 1
+ except Exception as e:
+ # A run that finished meanwhile cannot be cancelled; that is the
outcome wanted anyway.
+ # gh reports the status as "(HTTP 409)"; a bare "409" could be
part of the run id.
+ if "(HTTP 409)" in str(e):
+ continue
+ problems.append((merged.number, f"cancelling run {run.id}
({run.workflow}) failed: {e}"))
+ verb = "Would cancel" if client.dry_run else "Cancelled"
+ console_print(f"[info]{verb} {cancelled} runs of merged pull requests[/]")
+ return problems
diff --git a/dev/breeze/tests/test_fork_ci_gate.py
b/dev/breeze/tests/test_fork_ci_gate.py
index b52a0d29a1c..e641e45ca82 100644
--- a/dev/breeze/tests/test_fork_ci_gate.py
+++ b/dev/breeze/tests/test_fork_ci_gate.py
@@ -321,12 +321,17 @@ def test_parse_config_defaults():
assert config.allowlist == {}
assert config.pilot == {}
assert config.job_statuses is True
+ assert config.cancel_merged_runs is True
def test_parse_config_turns_job_statuses_off():
assert parse_config("job_statuses = false\n").job_statuses is False
+def test_parse_config_turns_cancelling_merged_runs_off():
+ assert parse_config("cancel_merged_runs = false\n").cancel_merged_runs is
False
+
+
def test_parse_config_lowercases_pilot_logins():
config = parse_config('[pilot]\n"Some-User" = 2026-10-20T00:00:00Z\n')
@@ -340,6 +345,9 @@ def test_parse_config_lowercases_pilot_logins():
[
pytest.param("enabled = 1\n", "enabled must be",
id="enabled-not-bool"),
pytest.param('job_statuses = "no"\n', "job_statuses must be",
id="job-statuses-not-bool"),
+ pytest.param(
+ "cancel_merged_runs = 0\n", "cancel_merged_runs must be",
id="cancel-merged-runs-not-bool"
+ ),
pytest.param(
"enabled_since = 2026-01-01T00:00:00\n", "enabled_since must be",
id="enabled-since-naive"
),
diff --git a/dev/breeze/tests/test_fork_ci_github.py
b/dev/breeze/tests/test_fork_ci_github.py
index d993add8bb5..09257d0697c 100644
--- a/dev/breeze/tests/test_fork_ci_github.py
+++ b/dev/breeze/tests/test_fork_ci_github.py
@@ -128,6 +128,7 @@ def test_dry_run_does_not_mutate(mock_run):
client.close(1)
client.remove_label(1, "fork ci required")
client.delete_run(1, 7)
+ client.cancel_run(1, 7)
mock_run.assert_not_called()
@@ -170,3 +171,19 @@ 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
+
+
[email protected](("force", "action"), [(False, "cancel"), (True,
"force-cancel")])
[email protected]("airflow_breeze.utils.fork_ci_github.subprocess.run",
autospec=True)
+def test_cancel_run(mock_run, force, action):
+ mock_run.return_value = _completed()
+
+ GitHubClient().cancel_run(1, 7, force=force)
+
+ assert mock_run.call_args.args[0] == [
+ "gh",
+ "api",
+ "-X",
+ "POST",
+ f"repos/apache/airflow/actions/runs/7/{action}",
+ ]
diff --git a/dev/breeze/tests/test_fork_ci_reconcile.py
b/dev/breeze/tests/test_fork_ci_reconcile.py
index 9f3aa61f751..3df786d4a83 100644
--- a/dev/breeze/tests/test_fork_ci_reconcile.py
+++ b/dev/breeze/tests/test_fork_ci_reconcile.py
@@ -2073,3 +2073,43 @@ def
test_job_status_reporter_compares_contexts_case_insensitively(client):
assert JobStatusReporter(client, repo=REPO).report(_pr(), COMPLETED_RUN)
== "1:1:FAILURE"
client.set_status.assert_not_called()
+
+
[email protected]("enabled", [True, False])
[email protected]("airflow_breeze.utils.fork_ci_reconcile.cancel_merged_pull_request_runs",
autospec=True)
+def test_run_reconcile_cancels_runs_of_merged_pull_requests(mock_cancel,
client, tmp_path, enabled):
+ mock_cancel.return_value = [(10, "cancelling run 7 failed")]
+ client.graphql.side_effect = [_search_result(), _search_result(),
_search_result()]
+ client.rest_get_pages.return_value = [{"total_count": 0, "workflow_runs":
[]}]
+
+ has_alerts = run_reconcile(
+ client,
+ dataclasses.replace(CONFIG, cancel_merged_runs=enabled),
+ now=NOW,
+ run_url="https://run";,
+ output_file=tmp_path / "s.json",
+ workflows_dir=tmp_path,
+ )
+
+ assert mock_cancel.called is enabled
+ assert has_alerts is enabled
+
+
[email protected]("airflow_breeze.utils.fork_ci_reconcile.cancel_merged_pull_request_runs",
autospec=True)
+def test_run_reconcile_reports_a_failed_cancellation_pass(mock_cancel, client,
tmp_path):
+ mock_cancel.side_effect = GitHubError("HTTP 502")
+ client.graphql.side_effect = [_search_result(), _search_result(),
_search_result()]
+ client.rest_get_pages.return_value = [{"total_count": 0, "workflow_runs":
[]}]
+
+ assert (
+ run_reconcile(
+ client,
+ CONFIG,
+ now=NOW,
+ run_url="https://run";,
+ output_file=tmp_path / "s.json",
+ workflows_dir=tmp_path,
+ )
+ is True
+ )
+ assert "cancelling runs of merged pull requests failed" in (tmp_path /
"s.json").read_text()
diff --git a/dev/breeze/tests/test_merged_pull_request_runs.py
b/dev/breeze/tests/test_merged_pull_request_runs.py
new file mode 100644
index 00000000000..5c6223963c0
--- /dev/null
+++ b/dev/breeze/tests/test_merged_pull_request_runs.py
@@ -0,0 +1,266 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+from __future__ import annotations
+
+from datetime import UTC, datetime, timedelta
+from unittest import mock
+
+import pytest
+
+from airflow_breeze.utils.fork_ci_github import GitHubClient, GitHubError
+from airflow_breeze.utils.merged_pull_request_runs import (
+ BRANCH_BATCH_SIZE,
+ PullRequestHead,
+ UnfinishedRun,
+ cancel_merged_pull_request_runs,
+ find_merged_pull_request,
+ find_pull_requests_by_branch,
+)
+
+REPO = "apache/airflow"
+NOW = datetime(2026, 10, 25, 19, 0, tzinfo=UTC)
+MERGED_AT = NOW - timedelta(minutes=10)
+RUN = UnfinishedRun(
+ id=7,
+ workflow="Tests (AMD)",
+ status="in_progress",
+ head_repo="someone/airflow",
+ head_branch="fix",
+ head_sha="aaa",
+ created_at=MERGED_AT - timedelta(minutes=20),
+)
+MERGED = PullRequestHead(
+ number=10, state="MERGED", head_repo="someone/airflow", head_sha="aaa",
merged_at=MERGED_AT
+)
+
+
[email protected]
+def client():
+ github = mock.create_autospec(GitHubClient, instance=True)
+ github.repo = REPO
+ github.dry_run = False
+ return github
+
+
[email protected](
+ ("pull_requests", "expected"),
+ [
+ pytest.param([MERGED], MERGED, id="merged-pull-request"),
+ pytest.param(
+ [PullRequestHead(10, "MERGED", "Someone/Airflow", "aaa",
MERGED_AT)],
+ PullRequestHead(10, "MERGED", "Someone/Airflow", "aaa", MERGED_AT),
+ id="head-repository-in-another-case",
+ ),
+ pytest.param(
+ [PullRequestHead(11, "OPEN", "someone/airflow", "aaa", None),
MERGED],
+ None,
+ id="open-pr-same-head",
+ ),
+ pytest.param(
+ [PullRequestHead(10, "MERGED", "other/airflow", "aaa",
MERGED_AT)], None, id="other-fork"
+ ),
+ pytest.param(
+ [PullRequestHead(10, "MERGED", "someone/airflow", "bbb",
MERGED_AT)], None, id="other-commit"
+ ),
+ pytest.param(
+ [PullRequestHead(10, "MERGED", "someone/airflow", "aaa",
RUN.created_at - timedelta(minutes=1))],
+ None,
+ id="run-started-after-the-merge",
+ ),
+ pytest.param(
+ [PullRequestHead(10, "CLOSED", "someone/airflow", "aaa", None)],
None, id="closed-not-merged"
+ ),
+ pytest.param([], None, id="no-pull-request"),
+ ],
+)
+def test_find_merged_pull_request(pull_requests, expected):
+ assert find_merged_pull_request(RUN, pull_requests) == expected
+
+
+def _run(run_id, *, status="in_progress", branch="fix", sha="aaa",
created_at=RUN.created_at):
+ return {
+ "id": run_id,
+ "name": "Tests (AMD)",
+ "status": status,
+ "head_repository": {"full_name": "someone/airflow"},
+ "head_branch": branch,
+ "head_sha": sha,
+ "created_at": created_at.strftime("%Y-%m-%dT%H:%M:%SZ"),
+ }
+
+
+def _pr_node(number, state, sha="aaa", merged_at=MERGED_AT):
+ return {
+ "number": number,
+ "state": state,
+ "mergedAt": merged_at.strftime("%Y-%m-%dT%H:%M:%SZ") if merged_at else
None,
+ "headRefOid": sha,
+ "headRepository": {"nameWithOwner": "someone/airflow"},
+ }
+
+
+def _runs_by_status(runs):
+ def pages(path):
+ status = path.split("status=", 1)[1].split("&", 1)[0]
+ return [{"workflow_runs": [run for run in runs if run["status"] ==
status]}]
+
+ return pages
+
+
[email protected](
+ ("merged_minutes_ago", "status", "force"),
+ [
+ pytest.param(10, "in_progress", False, id="recently-merged"),
+ pytest.param(31, "in_progress", True,
id="still-running-long-after-the-merge"),
+ pytest.param(31, "queued", False, id="queued-long-after-the-merge"),
+ ],
+)
+def test_cancel_merged_pull_request_runs(client, merged_minutes_ago, status,
force):
+ merged_at = NOW - timedelta(minutes=merged_minutes_ago)
+ client.rest_get_pages.side_effect = _runs_by_status(
+ [_run(7, status=status, created_at=merged_at - timedelta(minutes=5))]
+ )
+ client.graphql.return_value = {
+ "repository": {"b0": {"nodes": [_pr_node(10, "MERGED",
merged_at=merged_at)]}}
+ }
+
+ assert cancel_merged_pull_request_runs(client, repo=REPO, now=NOW) == []
+
+ client.cancel_run.assert_called_once_with(10, 7, force=force)
+
+
+def
test_cancel_merged_pull_request_runs_leaves_open_pull_requests_alone(client):
+ client.rest_get_pages.side_effect = _runs_by_status([_run(7)])
+ client.graphql.return_value = {"repository": {"b0": {"nodes":
[_pr_node(10, "OPEN", merged_at=None)]}}}
+
+ cancel_merged_pull_request_runs(client, repo=REPO, now=NOW)
+
+ client.cancel_run.assert_not_called()
+
+
+def test_cancel_merged_pull_request_runs_without_unfinished_runs(client):
+ client.rest_get_pages.side_effect = _runs_by_status([])
+
+ assert cancel_merged_pull_request_runs(client, repo=REPO, now=NOW) == []
+ client.graphql.assert_not_called()
+
+
[email protected](
+ ("error", "problems"),
+ [
+ pytest.param(
+ GitHubError(
+ "gh api -X POST repos/apache/airflow/actions/runs/7/cancel
failed: gh: done (HTTP 409)"
+ ),
+ [],
+ id="finished-meanwhile",
+ ),
+ pytest.param(
+ GitHubError("gh api repos/apache/airflow/actions/runs/1409/cancel
failed: HTTP 500"),
+ [
+ (
+ 10,
+ "cancelling run 7 (Tests (AMD)) failed: gh api
repos/apache/airflow/actions/runs/1409/cancel failed: HTTP 500",
+ )
+ ],
+ id="failed-with-409-in-the-path",
+ ),
+ ],
+)
+def test_cancel_merged_pull_request_runs_reports_failed_cancellations(client,
error, problems):
+ client.rest_get_pages.side_effect = _runs_by_status([_run(7), _run(8)])
+ client.graphql.return_value = {"repository": {"b0": {"nodes":
[_pr_node(10, "MERGED")]}}}
+ client.cancel_run.side_effect = [error, None]
+
+ assert cancel_merged_pull_request_runs(client, repo=REPO, now=NOW) ==
problems
+ assert client.cancel_run.call_count == 2
+
+
+def test_find_pull_requests_by_branch_batches_branches(client):
+ branches = [f"branch-{index:02d}" for index in range(BRANCH_BATCH_SIZE +
1)]
+
+ def graphql(query, variables):
+ batch = sorted(key for key in variables if key.startswith("b"))
+ return {"repository": {key: {"nodes": [_pr_node(1, "MERGED")]} for key
in batch}}
+
+ client.graphql.side_effect = graphql
+
+ found = find_pull_requests_by_branch(client, REPO, [*branches,
branches[0]])
+
+ assert sorted(found) == branches
+ assert client.graphql.call_count == 2
+ assert client.graphql.call_args_list[0].args[1]["owner"] == "apache"
+
+
+def test_find_pull_requests_by_branch_reads_every_page(client):
+ first = {"pageInfo": {"hasNextPage": True, "endCursor": "c1"}, "nodes":
[_pr_node(1, "MERGED")]}
+ second = {
+ "pageInfo": {"hasNextPage": False, "endCursor": None},
+ "nodes": [_pr_node(2, "OPEN", merged_at=None)],
+ }
+ client.graphql.side_effect = [
+ {"repository": {"b0": first}},
+ {"repository": {"pullRequests": second}},
+ ]
+
+ found = find_pull_requests_by_branch(client, REPO, ["main"])
+
+ assert [pr.number for pr in found["main"]] == [1, 2]
+ assert client.graphql.call_args.args[1] == {
+ "owner": "apache",
+ "name": "airflow",
+ "branch": "main",
+ "cursor": "c1",
+ }
+
+
+def test_open_pull_request_on_a_later_page_keeps_the_run(client):
+ client.rest_get_pages.side_effect = _runs_by_status([_run(7)])
+ client.graphql.side_effect = [
+ {
+ "repository": {
+ "b0": {
+ "pageInfo": {"hasNextPage": True, "endCursor": "c1"},
+ "nodes": [_pr_node(10, "MERGED")],
+ }
+ }
+ },
+ {
+ "repository": {
+ "pullRequests": {
+ "pageInfo": {"hasNextPage": False},
+ "nodes": [_pr_node(11, "OPEN", merged_at=None)],
+ }
+ }
+ },
+ ]
+
+ cancel_merged_pull_request_runs(client, repo=REPO, now=NOW)
+
+ client.cancel_run.assert_not_called()
+
+
[email protected](("dry_run", "summary"), [(False, "Cancelled 1 runs"),
(True, "Would cancel 1 runs")])
[email protected]("airflow_breeze.utils.merged_pull_request_runs.console_print",
autospec=True)
+def test_cancel_merged_pull_request_runs_summary_tells_a_dry_run(mock_print,
client, dry_run, summary):
+ client.dry_run = dry_run
+ client.rest_get_pages.side_effect = _runs_by_status([_run(7)])
+ client.graphql.return_value = {"repository": {"b0": {"nodes":
[_pr_node(10, "MERGED")]}}}
+
+ cancel_merged_pull_request_runs(client, repo=REPO, now=NOW)
+
+ assert summary in mock_print.call_args.args[0]
