This is an automated email from the ASF dual-hosted git repository.
potiuk pushed a commit to branch v3-4-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/v3-4-test by this push:
new ee375024ffd [v3-4-test] Cancel the runs of merged pull requests when
they merge (#75683) (#75687)
ee375024ffd is described below
commit ee375024ffd5fa5418f84d05e8fd37543b7df5e3
Author: github-actions[bot]
<41898282+github-actions[bot]@users.noreply.github.com>
AuthorDate: Mon Oct 26 20:25:38 2026 +0100
[v3-4-test] Cancel the runs of merged pull requests when they merge
(#75683) (#75687)
The fork CI monitor cancelled them on its 15-minute schedule, so runs of a
merged pull request kept holding runners for up to 15 minutes, and stuck
ones for 30 more before being force-cancelled. Merging pushes to the target
branch, which lets a push-triggered workflow find the merged pull request
directly and cancel its runs right away, without the monitor's search over
every unfinished run.
(cherry picked from commit bfb620d271b0dca8f3a0f42966603b7290273236)
Generated-by: Claude Opus 5
Co-authored-by: Jarek Potiuk <[email protected]>
---
.github/fork-ci-gate.toml | 5 -
.github/workflows/cancel-merged-pr-runs.yml | 119 +++++++++
.github/workflows/fork-ci-monitor.yml | 4 +-
dev/breeze/doc/08_ci_tasks.rst | 3 +-
dev/breeze/doc/ci/05_workflows.md | 25 ++
.../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 ---------------------
14 files changed, 148 insertions(+), 604 deletions(-)
diff --git a/.github/fork-ci-gate.toml b/.github/fork-ci-gate.toml
index 6119f610ef2..6fc7a0de1ce 100644
--- a/.github/fork-ci-gate.toml
+++ b/.github/fork-ci-gate.toml
@@ -34,11 +34,6 @@ 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/cancel-merged-pr-runs.yml
b/.github/workflows/cancel-merged-pr-runs.yml
new file mode 100644
index 00000000000..72c28642baa
--- /dev/null
+++ b/.github/workflows/cancel-merged-pr-runs.yml
@@ -0,0 +1,119 @@
+# 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 `pull_request` runs that are still going after their pull
request merged. Merging
+# pushes to the target branch, so this starts right after the merge. A push
run can cancel the runs
+# of fork pull requests too. It never checks out or runs pull request code.
See "Cancelling runs of
+# merged pull requests" in dev/breeze/doc/ci/05_workflows.md.
+name: "Cancel the runs of merged pull requests"
+on: # yamllint disable-line rule:truthy
+ push:
+ # The branches pull requests merge into, as in the `pull_request` trigger
of ci-amd.yml. A push
+ # runs the workflow file of the pushed branch, so a release branch needs
its own copy.
+ branches:
+ - main
+ - v[0-9]+-[0-9]+-test
+ - v[0-9]+-[0-9]+-stable
+ - providers-[a-z]+-?[a-z]*/v[0-9]+-[0-9]+
+ - chart/v[0-9]+-[0-9]+x-test
+ - chart/v[0-9]+-[0-9]+x-stable
+ - airflow-ctl/v[0-9]+-[0-9]+-test
+ - airflow-ctl/v[0-9]+-[0-9]+-stable
+permissions:
+ actions: write
+ contents: read
+ pull-requests: read
+jobs:
+ cancel:
+ name: "Cancel the runs of merged pull requests"
+ if: github.repository == 'apache/airflow'
+ runs-on: ["ubuntu-slim"]
+ timeout-minutes: 10
+ steps:
+ - name: "Cancel the runs"
+ env:
+ GH_TOKEN: ${{ secrets.GITHUB_TOKEN }}
+ REPO: ${{ github.repository }}
+ SHA: ${{ github.sha }}
+ run: |
+ # The statuses of a run that has not finished. A run waiting for
approval is "completed".
+ statuses=(queued in_progress waiting pending requested)
+ # Runs of a fork pull request do not list their pull request, so
runs are matched by the
+ # head commit and repository. Only runs created before the merge
count: workflows
+ # triggered by the merge itself, for example on `closed`, are meant
to run.
+ list_runs() {
+ local head_repo=$1 head_sha=$2 merged_at=$3 status
+ for status in "${statuses[@]}"; do
+ gh api --paginate \
+
"repos/${REPO}/actions/runs?event=pull_request&head_sha=${head_sha}&status=${status}"
\
+ --jq ".workflow_runs[] | select(.head_repository.full_name ==
\"${head_repo}\"
+ and .created_at < \"${merged_at}\") | .id" || return 1
+ done
+ }
+ # Lookups run in command substitutions rather than process
substitutions, so that a failed
+ # one fails the job instead of looking like nothing to cancel.
GitHub associates the merge
+ # commit with its pull request some seconds after the push, as
automatic-backport.yml
+ # also notes; a push that is not a merge finds nothing after a
minute.
+ for _ in {1..6}; do
+ merged=$(gh api "repos/${REPO}/commits/${SHA}/pulls" --jq '.[]
+ | select(.merged_at != null)
+ | [.number, .head.repo.full_name, .head.sha, .merged_at] | @tsv')
+ [[ -n "${merged}" ]] && break
+ sleep 10
+ done
+ to_cancel=()
+ while IFS=$'\t' read -r number head_repo head_sha merged_at; do
+ [[ -n "${number}" ]] || continue
+ # The same commit can be the head of another pull request that is
still open.
+ open=$(gh pr list -R "${REPO}" --state open --search "${head_sha}"
--json headRefOid \
+ --jq "[.[] | select(.headRefOid == \"${head_sha}\")] | length")
+ if [[ "${open}" != 0 ]]; then
+ echo "#${number}: ${head_sha} is the head of an open pull
request, nothing cancelled"
+ continue
+ fi
+ runs=$(list_runs "${head_repo}" "${head_sha}" "${merged_at}")
+ mapfile -t ids <<< "${runs}"
+ for id in "${ids[@]}"; do
+ [[ -n "${id}" ]] || continue
+ echo "#${number}: cancel run ${id}"
+ # A run that finished meanwhile cannot be cancelled (HTTP 409);
that is fine.
+ gh api -X POST "repos/${REPO}/actions/runs/${id}/cancel" >
/dev/null || true
+ to_cancel+=("${id}")
+ done
+ done <<< "${merged}"
+ # A job that ignores the cancellation keeps its runner, so a run
still going after five
+ # minutes is force-cancelled.
+ for _ in {1..10}; do
+ remaining=()
+ for id in "${to_cancel[@]}"; do
+ [[ "$(gh api "repos/${REPO}/actions/runs/${id}" --jq .status)"
== completed ]] \
+ || remaining+=("${id}")
+ done
+ to_cancel=("${remaining[@]}")
+ [[ ${#to_cancel[@]} == 0 ]] && exit 0
+ sleep 30
+ done
+ failed=0
+ for id in "${to_cancel[@]}"; do
+ echo "force-cancel run ${id}"
+ if ! out=$(gh api -X POST
"repos/${REPO}/actions/runs/${id}/force-cancel" 2>&1); then
+ echo "${out}"
+ [[ "${out}" == *"(HTTP 409)"* ]] || failed=1
+ fi
+ done
+ exit "${failed}"
diff --git a/.github/workflows/fork-ci-monitor.yml
b/.github/workflows/fork-ci-monitor.yml
index 278787be5af..ec34b9615b0 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, deleting the
runs that only gated a
- # pull request to fork CI, and cancelling the unfinished runs of merged
pull requests.
+ # Listing the runs and jobs of the gate tampering check, and deleting
the runs that only
+ # gated a pull request to fork CI.
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 7d00e36ee32..d54eadbaa69 100644
--- a/dev/breeze/doc/08_ci_tasks.rst
+++ b/dev/breeze/doc/08_ci_tasks.rst
@@ -188,8 +188,7 @@ 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. It also cancels
-the unfinished workflow runs of pull requests, of any author, that have
already merged. With
+by changing workflows, and writes everything it could not classify to a Slack
message. 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/05_workflows.md
b/dev/breeze/doc/ci/05_workflows.md
index 1760f529690..264bf7378e6 100644
--- a/dev/breeze/doc/ci/05_workflows.md
+++ b/dev/breeze/doc/ci/05_workflows.md
@@ -244,6 +244,31 @@ Each composite workflow is stored in `.github/workflows/`
and gets called using
GitHub Actions includes a concurrency feature that we use to prevent wasted
resources. If you push new commits while tests are already running, the system
cancels the old test run and starts over with your latest code. This saves time
and compute resources.
+### Cancelling runs of merged pull requests
+
+Merging a pull request does not stop its `pull_request` runs, and the
concurrency group does not
+help, because the merge starts no new run in the pull request's group.
Maintainers often merge
+once the checks they need are green, while slower jobs are still queued or
running for a result
+nobody reads.
+
+Merging pushes to the target branch, which starts `cancel-merged-pr-runs.yml`.
It looks up the
+merged pull request of the pushed commit and cancels its `pull_request` runs
that are `queued`,
+`in_progress`, `waiting`, `pending` or `requested`. A push run has a token
that can write to
+`apache/airflow`, so it cancels the runs of fork pull requests too. A run is
cancelled when:
+
+- its head repository and commit are the merged pull request's (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 commit; such runs are left to finish,
since no later push
+ revisits them;
+- it was created before the merge. A workflow triggered by the merge itself,
such as one on the
+ `closed` event, is meant to run.
+
+A run still not finished five minutes after the cancel, for example because a
step ignores the
+cancellation, is force-cancelled. A run of an earlier commit of the pull
request is normally
+cancelled by its concurrency group when the next commit is pushed. A push runs
the workflow file
+of the pushed branch, so the cancellation works on a release branch only once
the workflow is on
+that branch.
+
## Branch-Specific Behavior
Different branches in our repository serve different purposes, so they get
different levels of testing.
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 004c097be91..6335400d128 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
@@ -833,36 +833,6 @@ 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
@@ -917,8 +887,7 @@ 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. 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
+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
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 4816e81bfd3..10fcc9a106a 100644
--- a/dev/breeze/src/airflow_breeze/utils/fork_ci_gate.py
+++ b/dev/breeze/src/airflow_breeze/utils/fork_ci_gate.py
@@ -129,7 +129,6 @@ 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()
@@ -178,9 +177,6 @@ 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,
@@ -188,7 +184,6 @@ 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 fe3fdfb651e..73a44503acb 100644
--- a/dev/breeze/src/airflow_breeze/utils/fork_ci_github.py
+++ b/dev/breeze/src/airflow_breeze/utils/fork_ci_github.py
@@ -172,14 +172,6 @@ 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 00698b78494..6b54f6c7fac 100644
--- a/dev/breeze/src/airflow_breeze/utils/fork_ci_reconcile.py
+++ b/dev/breeze/src/airflow_breeze/utils/fork_ci_reconcile.py
@@ -74,7 +74,6 @@ 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.
@@ -1548,13 +1547,6 @@ 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
deleted file mode 100644
index 7eed315630b..00000000000
--- a/dev/breeze/src/airflow_breeze/utils/merged_pull_request_runs.py
+++ /dev/null
@@ -1,211 +0,0 @@
-# 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 ca4194b99d2..51f351e8b03 100644
--- a/dev/breeze/tests/test_fork_ci_gate.py
+++ b/dev/breeze/tests/test_fork_ci_gate.py
@@ -321,17 +321,12 @@ 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')
@@ -345,9 +340,6 @@ 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 09257d0697c..d993add8bb5 100644
--- a/dev/breeze/tests/test_fork_ci_github.py
+++ b/dev/breeze/tests/test_fork_ci_github.py
@@ -128,7 +128,6 @@ 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()
@@ -171,19 +170,3 @@ 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 3df786d4a83..9f3aa61f751 100644
--- a/dev/breeze/tests/test_fork_ci_reconcile.py
+++ b/dev/breeze/tests/test_fork_ci_reconcile.py
@@ -2073,43 +2073,3 @@ 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
deleted file mode 100644
index 5c6223963c0..00000000000
--- a/dev/breeze/tests/test_merged_pull_request_runs.py
+++ /dev/null
@@ -1,266 +0,0 @@
-# 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]
