Cancel the workflow runs of pull requests that have already merged

.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]

.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.

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.
 

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

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,13 +178,17 @@ 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,
         close_after_days=close_after_days,
         allowlist=allowlist,
         pilot=pilot,
         job_statuses=job_statuses,
+        cancel_merged_runs=cancel_merged_runs,
     )
 
 

dev/breeze/src/airflow_breeze/utils/fork_ci_github.py

@@ -172,6 +172,14 @@ def delete_run(self, pr: int, run_id: int) -> None:
             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,

dev/breeze/src/airflow_breeze/utils/fork_ci_reconcile.py

@@ -74,6 +74,7 @@
     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: "

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

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"
         ),