From c2a678be66e858b80f7127d4c36dcab857ca2e32 Mon Sep 17 00:00:00 2001 From: XuPeng-SH Date: Fri, 4 Sep 2026 00:56:34 +0800 Subject: [PATCH 1/3] ci: cancel stale pull request runs --- .github/workflows/pr-title.yml | 2 +- .github/workflows/static-checks.yml | 4 +- .github/workflows/supersede-pr-runs.yml | 40 +++++ .github/workflows/test.yml | 2 +- CONTRIBUTING.md | 8 + scripts/ci/cancel_stale_pr_runs.py | 219 ++++++++++++++++++++++++ scripts/ci/test_cancel_stale_pr_runs.py | 212 +++++++++++++++++++++++ 7 files changed, 483 insertions(+), 4 deletions(-) create mode 100644 .github/workflows/supersede-pr-runs.yml create mode 100755 scripts/ci/cancel_stale_pr_runs.py create mode 100755 scripts/ci/test_cancel_stale_pr_runs.py diff --git a/.github/workflows/pr-title.yml b/.github/workflows/pr-title.yml index 32aa594cf4..0268b9e77c 100644 --- a/.github/workflows/pr-title.yml +++ b/.github/workflows/pr-title.yml @@ -9,7 +9,7 @@ on: types: [opened, edited, synchronize] concurrency: - group: ${{ github.workflow }}-${{ github.head_ref }} + group: ${{ github.workflow }}-${{ github.event.pull_request.number }} cancel-in-progress: true jobs: diff --git a/.github/workflows/static-checks.yml b/.github/workflows/static-checks.yml index a9af7f502d..3ff286308d 100644 --- a/.github/workflows/static-checks.yml +++ b/.github/workflows/static-checks.yml @@ -10,7 +10,7 @@ on: branches: [main, develop] concurrency: - group: ${{ github.workflow }}-${{ github.head_ref || github.ref }} + group: ${{ github.workflow }}-${{ github.event.pull_request.number || github.ref }} cancel-in-progress: true env: @@ -51,7 +51,7 @@ jobs: - name: Validate repository metadata and documentation run: python3 scripts/ci/validate_repository.py - name: Test CI scope routing - run: python3 scripts/ci/test_ci_scope.py + run: python3 -m unittest discover -s scripts/ci -p 'test_*.py' - name: Validate Terminal-Bench harness contracts if: ${{ needs.scope.result != 'success' || needs.scope.outputs.harness == 'true' }} run: | diff --git a/.github/workflows/supersede-pr-runs.yml b/.github/workflows/supersede-pr-runs.yml new file mode 100644 index 0000000000..7f7554c9da --- /dev/null +++ b/.github/workflows/supersede-pr-runs.yml @@ -0,0 +1,40 @@ +name: Supersede stale PR runs + +on: + pull_request_target: + branches: [main, develop] + types: [synchronize] + +# This controller runs from the trusted base revision. It must never check out +# or execute the pull request head while holding permission to cancel runs. +permissions: + actions: write + contents: read + pull-requests: read + +concurrency: + group: supersede-pr-runs-${{ github.event.pull_request.number }} + cancel-in-progress: true + +jobs: + cancel-stale-runs: + name: Cancel superseded PR runs + runs-on: ubuntu-latest + timeout-minutes: 5 + steps: + - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7 + with: + ref: ${{ github.event.pull_request.base.sha }} + fetch-depth: 1 + persist-credentials: false + - name: Cancel runs for earlier PR heads + env: + GH_TOKEN: ${{ github.token }} + EVENT_HEAD_SHA: ${{ github.event.pull_request.head.sha }} + PR_NUMBER: ${{ github.event.pull_request.number }} + REPOSITORY: ${{ github.repository }} + run: >- + python3 scripts/ci/cancel_stale_pr_runs.py + --repository "$REPOSITORY" + --pr-number "$PR_NUMBER" + --event-head-sha "$EVENT_HEAD_SHA" diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml index f85880dc3c..d71cc5bb8f 100644 --- a/.github/workflows/test.yml +++ b/.github/workflows/test.yml @@ -10,7 +10,7 @@ on: branches: [main, develop] concurrency: - group: ${{ github.workflow }}-${{ github.head_ref || github.ref }} + group: ${{ github.workflow }}-${{ github.event.pull_request.number || github.ref }} cancel-in-progress: true env: diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index e2f70322fd..a1c9360cb6 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -87,6 +87,14 @@ Required check names remain present when their heavy work is skipped, so this routing is compatible with branch protection and the merge queue. The routing contract and its tests live in [`scripts/ci/`](scripts/ci/). +For fork pull requests, an update may require maintainer approval before the +replacement test run can enter the normal concurrency group. A separate +base-revision controller cancels active runs for earlier heads immediately. The +controller never checks out or executes pull request code; the test workflows +remain low-privilege `pull_request` workflows. Normal concurrency groups use the +pull request number, so identically named branches in different forks remain +isolated. + ## Open a pull request 1. Rebase the feature branch on the current `main` branch. diff --git a/scripts/ci/cancel_stale_pr_runs.py b/scripts/ci/cancel_stale_pr_runs.py new file mode 100755 index 0000000000..e94058c273 --- /dev/null +++ b/scripts/ci/cancel_stale_pr_runs.py @@ -0,0 +1,219 @@ +#!/usr/bin/env python3 +"""Cancel active GitHub Actions runs that belong to an earlier head of a PR.""" + +from __future__ import annotations + +import argparse +from dataclasses import dataclass +import json +import os +import re +import sys +from typing import Any, Protocol +from urllib.error import HTTPError +from urllib.parse import urlencode, urljoin, urlsplit +from urllib.request import Request, urlopen + + +ACTIVE_RUN_STATUSES = ("queued", "in_progress", "pending", "requested", "waiting") +MAX_PAGES_PER_STATUS = 10 +MAX_STALE_RUNS = 1_000 +REPOSITORY_PATTERN = re.compile(r"^[A-Za-z0-9_.-]+/[A-Za-z0-9_.-]+$") +SHA_PATTERN = re.compile(r"^[0-9a-fA-F]{40}$") + + +@dataclass(frozen=True) +class PullRequestHead: + repository_id: int + repository: str + ref: str + sha: str + + +class ActionsApi(Protocol): + def get_pull_request(self, repository: str, number: int) -> dict[str, Any]: ... + + def list_runs(self, repository: str, status: str) -> list[dict[str, Any]]: ... + + def cancel_run(self, repository: str, run_id: int) -> bool: ... + + +class GitHubApi: + """Small GitHub REST client with bounded, origin-checked pagination.""" + + def __init__(self, token: str, api_url: str = "https://api.github.com") -> None: + if not token: + raise ValueError("GH_TOKEN is required") + self._token = token + self._api_url = api_url.rstrip("/") + "/" + self._api_origin = urlsplit(self._api_url).netloc + + def _request(self, method: str, url: str) -> tuple[bytes, Any]: + target = url if urlsplit(url).scheme else urljoin(self._api_url, url.lstrip("/")) + if urlsplit(target).netloc != self._api_origin: + raise RuntimeError("refusing to send the GitHub token to another origin") + request = Request( + target, + method=method, + headers={ + "Accept": "application/vnd.github+json", + "Authorization": f"Bearer {self._token}", + "User-Agent": "astra-stale-pr-run-controller", + "X-GitHub-Api-Version": "2022-11-28", + }, + ) + with urlopen(request, timeout=30) as response: + return response.read(), response.headers + + def _get_json(self, path: str) -> dict[str, Any]: + body, _ = self._request("GET", path) + value = json.loads(body) + if not isinstance(value, dict): + raise RuntimeError(f"GitHub returned a non-object response for {path}") + return value + + @staticmethod + def _next_link(headers: Any) -> str | None: + for part in headers.get("Link", "").split(","): + match = re.match(r'\s*<([^>]+)>;\s*rel="([^"]+)"\s*$', part) + if match and match.group(2) == "next": + return match.group(1) + return None + + def get_pull_request(self, repository: str, number: int) -> dict[str, Any]: + return self._get_json(f"repos/{repository}/pulls/{number}") + + def list_runs(self, repository: str, status: str) -> list[dict[str, Any]]: + query = urlencode({"event": "pull_request", "status": status, "per_page": 100}) + next_url: str | None = f"repos/{repository}/actions/runs?{query}" + runs: list[dict[str, Any]] = [] + visited: set[str] = set() + while next_url: + if next_url in visited or len(visited) >= MAX_PAGES_PER_STATUS: + raise RuntimeError(f"workflow-run pagination exceeded its safe bound for {status}") + visited.add(next_url) + body, headers = self._request("GET", next_url) + page = json.loads(body) + page_runs = page.get("workflow_runs") if isinstance(page, dict) else None + if not isinstance(page_runs, list): + raise RuntimeError("GitHub workflow-runs response omitted workflow_runs") + runs.extend(run for run in page_runs if isinstance(run, dict)) + next_url = self._next_link(headers) + return runs + + def cancel_run(self, repository: str, run_id: int) -> bool: + try: + self._request("POST", f"repos/{repository}/actions/runs/{run_id}/cancel") + except HTTPError as error: + # A selected run can finish between the list and cancel requests. + if error.code == 409: + return False + raise + return True + + +def _pull_request_head(pull_request: dict[str, Any]) -> PullRequestHead: + head = pull_request.get("head") + if not isinstance(head, dict): + raise RuntimeError("pull request response omitted head") + repository = head.get("repo") + full_name = repository.get("full_name") if isinstance(repository, dict) else None + repository_id = repository.get("id") if isinstance(repository, dict) else None + ref = head.get("ref") + sha = head.get("sha") + if not isinstance(repository_id, int) or not all( + isinstance(value, str) and value for value in (full_name, ref, sha) + ): + raise RuntimeError("pull request response has an incomplete head identity") + return PullRequestHead(repository_id, full_name, ref, sha) + + +def select_stale_runs( + runs: list[dict[str, Any]], current_head: PullRequestHead +) -> list[dict[str, Any]]: + """Select prior-head runs for exactly the current fork repository and ref.""" + selected: dict[int, dict[str, Any]] = {} + for run in runs: + run_id = run.get("id") + head_repository = run.get("head_repository") + repository_id = head_repository.get("id") if isinstance(head_repository, dict) else None + if not isinstance(run_id, int): + continue + if repository_id != current_head.repository_id: + continue + if run.get("head_branch") != current_head.ref: + continue + if run.get("head_sha") == current_head.sha: + continue + selected[run_id] = run + return [selected[run_id] for run_id in sorted(selected)] + + +def cancel_stale_runs( + api: ActionsApi, repository: str, pr_number: int, event_head_sha: str +) -> tuple[int, int]: + pull_request = api.get_pull_request(repository, pr_number) + current_head = _pull_request_head(pull_request) + if pull_request.get("state") != "open" or current_head.sha != event_head_sha: + print( + "Ignoring a stale controller event: " + f"event head {event_head_sha}, live head {current_head.sha}, " + f"state {pull_request.get('state')}." + ) + return 0, 0 + + active_runs: list[dict[str, Any]] = [] + for status in ACTIVE_RUN_STATUSES: + active_runs.extend(api.list_runs(repository, status)) + stale_runs = select_stale_runs(active_runs, current_head) + if len(stale_runs) > MAX_STALE_RUNS: + raise RuntimeError( + f"refusing an unexpectedly large cancellation set ({len(stale_runs)} runs)" + ) + + cancelled = 0 + for run in stale_runs: + run_id = int(run["id"]) + if api.cancel_run(repository, run_id): + cancelled += 1 + print( + f"Cancellation requested for {run.get('name', 'workflow')} " + f"run {run_id} at {run.get('head_sha', '')}." + ) + else: + print(f"Run {run_id} became terminal before cancellation; no action needed.") + + print( + f"Found {len(stale_runs)} stale active run(s) for " + f"{current_head.repository}:{current_head.ref}; requested {cancelled} cancellation(s)." + ) + return len(stale_runs), cancelled + + +def parse_args(argv: list[str]) -> argparse.Namespace: + parser = argparse.ArgumentParser() + parser.add_argument("--repository", required=True) + parser.add_argument("--pr-number", required=True, type=int) + parser.add_argument("--event-head-sha", required=True) + args = parser.parse_args(argv) + if not REPOSITORY_PATTERN.fullmatch(args.repository): + parser.error("--repository must be an owner/repository pair") + if args.pr_number <= 0: + parser.error("--pr-number must be positive") + if not SHA_PATTERN.fullmatch(args.event_head_sha): + parser.error("--event-head-sha must be a 40-character hexadecimal commit SHA") + return args + + +def main(argv: list[str] | None = None) -> int: + args = parse_args(sys.argv[1:] if argv is None else argv) + api = GitHubApi( + os.environ.get("GH_TOKEN", ""), + os.environ.get("GITHUB_API_URL", "https://api.github.com"), + ) + cancel_stale_runs(api, args.repository, args.pr_number, args.event_head_sha) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/scripts/ci/test_cancel_stale_pr_runs.py b/scripts/ci/test_cancel_stale_pr_runs.py new file mode 100755 index 0000000000..3a24a4b517 --- /dev/null +++ b/scripts/ci/test_cancel_stale_pr_runs.py @@ -0,0 +1,212 @@ +#!/usr/bin/env python3 + +from __future__ import annotations + +from contextlib import redirect_stdout +from io import StringIO +import json +from pathlib import Path +import sys +import unittest +from unittest.mock import Mock + +sys.path.insert(0, str(Path(__file__).resolve().parent)) + +from cancel_stale_pr_runs import ( # noqa: E402 + ACTIVE_RUN_STATUSES, + GitHubApi, + MAX_STALE_RUNS, + PullRequestHead, + cancel_stale_runs, + select_stale_runs, +) + + +CURRENT_SHA = "b" * 40 +OLD_SHA = "a" * 40 + + +def workflow_run( + run_id: int, + *, + sha: str = OLD_SHA, + repository_id: int = 101, + branch: str = "fix/runtime", +) -> dict[str, object]: + return { + "id": run_id, + "name": "Test Suite", + "head_sha": sha, + "head_branch": branch, + "head_repository": {"id": repository_id, "full_name": "contributor/Astra"}, + } + + +class FakeApi: + def __init__(self, *, live_sha: str = CURRENT_SHA, state: str = "open") -> None: + self.pull_request = { + "state": state, + "head": { + "sha": live_sha, + "ref": "fix/runtime", + "repo": {"id": 101, "full_name": "contributor/Astra"}, + }, + } + self.runs_by_status: dict[str, list[dict[str, object]]] = { + status: [] for status in ACTIVE_RUN_STATUSES + } + self.cancelled: list[int] = [] + self.finished_before_cancel: set[int] = set() + + def get_pull_request(self, repository: str, number: int) -> dict[str, object]: + self.requested_pull = (repository, number) + return self.pull_request + + def list_runs(self, repository: str, status: str) -> list[dict[str, object]]: + return self.runs_by_status[status] + + def cancel_run(self, repository: str, run_id: int) -> bool: + if run_id in self.finished_before_cancel: + return False + self.cancelled.append(run_id) + return True + + +class SelectStaleRunsTests(unittest.TestCase): + def test_selects_only_prior_heads_of_the_exact_fork_ref(self) -> None: + head = PullRequestHead(101, "Contributor/Astra", "fix/runtime", CURRENT_SHA) + runs = [ + workflow_run(1), + workflow_run(2, sha=CURRENT_SHA), + workflow_run(3, repository_id=202), + workflow_run(4, branch="fix/other"), + {"id": 5, "head_sha": OLD_SHA, "head_branch": "fix/runtime"}, + ] + + self.assertEqual([run["id"] for run in select_stale_runs(runs, head)], [1]) + + def test_deduplicates_a_run_observed_during_status_transition(self) -> None: + head = PullRequestHead(101, "contributor/Astra", "fix/runtime", CURRENT_SHA) + run = workflow_run(7) + + self.assertEqual([item["id"] for item in select_stale_runs([run, run], head)], [7]) + + +class GitHubApiTests(unittest.TestCase): + def test_list_runs_follows_pagination(self) -> None: + api = GitHubApi("test-token") + second_page = "https://api.github.com/repos/matrixorigin/Astra/actions/runs?page=2" + api._request = Mock( # type: ignore[method-assign] + side_effect=[ + ( + json.dumps({"workflow_runs": [workflow_run(1)]}).encode(), + {"Link": f'<{second_page}>; rel="next"'}, + ), + (json.dumps({"workflow_runs": [workflow_run(2)]}).encode(), {}), + ] + ) + + self.assertEqual( + [run["id"] for run in api.list_runs("matrixorigin/Astra", "in_progress")], + [1, 2], + ) + + def test_request_rejects_a_foreign_origin_before_sending_the_token(self) -> None: + api = GitHubApi("test-token") + + with self.assertRaisesRegex(RuntimeError, "another origin"): + api._request("GET", "https://example.test/steal") + + +class CancelStaleRunsTests(unittest.TestCase): + def test_cancels_prior_runs_across_every_active_status(self) -> None: + api = FakeApi() + for index, status in enumerate(ACTIVE_RUN_STATUSES, start=1): + api.runs_by_status[status] = [workflow_run(index)] + api.runs_by_status["pending"].append(workflow_run(100, sha=CURRENT_SHA)) + + found, cancelled = cancel_stale_runs(api, "matrixorigin/Astra", 42, CURRENT_SHA) + + self.assertEqual( + (found, cancelled), + (len(ACTIVE_RUN_STATUSES), len(ACTIVE_RUN_STATUSES)), + ) + self.assertEqual(api.cancelled, list(range(1, len(ACTIVE_RUN_STATUSES) + 1))) + + def test_stale_controller_cannot_cancel_a_newer_head(self) -> None: + api = FakeApi(live_sha="c" * 40) + api.runs_by_status["in_progress"] = [workflow_run(1, sha=CURRENT_SHA)] + + found, cancelled = cancel_stale_runs(api, "matrixorigin/Astra", 42, CURRENT_SHA) + + self.assertEqual((found, cancelled), (0, 0)) + self.assertEqual(api.cancelled, []) + + def test_closed_pull_request_event_is_a_safe_noop(self) -> None: + api = FakeApi(state="closed") + api.runs_by_status["in_progress"] = [workflow_run(1)] + + found, cancelled = cancel_stale_runs(api, "matrixorigin/Astra", 42, CURRENT_SHA) + + self.assertEqual((found, cancelled), (0, 0)) + self.assertEqual(api.cancelled, []) + + def test_completion_race_is_not_reported_as_a_failed_cancellation(self) -> None: + api = FakeApi() + api.runs_by_status["in_progress"] = [workflow_run(1), workflow_run(2)] + api.finished_before_cancel.add(2) + + with redirect_stdout(StringIO()): + found, cancelled = cancel_stale_runs(api, "matrixorigin/Astra", 42, CURRENT_SHA) + + self.assertEqual((found, cancelled), (2, 1)) + self.assertEqual(api.cancelled, [1]) + + def test_unexpected_cancellation_fanout_fails_before_mutation(self) -> None: + api = FakeApi() + api.runs_by_status["queued"] = [ + workflow_run(run_id) for run_id in range(1, MAX_STALE_RUNS + 2) + ] + + with self.assertRaisesRegex(RuntimeError, "unexpectedly large"): + cancel_stale_runs(api, "matrixorigin/Astra", 42, CURRENT_SHA) + + self.assertEqual(api.cancelled, []) + + +class WorkflowSecurityContractTests(unittest.TestCase): + def test_controller_executes_only_the_trusted_base_revision(self) -> None: + workflow = ( + Path(__file__).resolve().parents[2] + / ".github/workflows/supersede-pr-runs.yml" + ).read_text(encoding="utf-8") + + for required in ( + "pull_request_target:", + "actions: write", + "contents: read", + "pull-requests: read", + "ref: ${{ github.event.pull_request.base.sha }}", + "persist-credentials: false", + "python3 scripts/ci/cancel_stale_pr_runs.py", + ): + with self.subTest(required=required): + self.assertIn(required, workflow) + self.assertNotIn("ref: ${{ github.event.pull_request.head.sha }}", workflow) + self.assertNotIn("secrets.", workflow) + + def test_pull_request_concurrency_isolated_by_pr_number(self) -> None: + root = Path(__file__).resolve().parents[2] + expected = "${{ github.event.pull_request.number" + for relative in ( + ".github/workflows/pr-title.yml", + ".github/workflows/static-checks.yml", + ".github/workflows/test.yml", + ".github/workflows/supersede-pr-runs.yml", + ): + with self.subTest(workflow=relative): + self.assertIn(expected, (root / relative).read_text(encoding="utf-8")) + + +if __name__ == "__main__": + unittest.main() From 9b59f331efbde3f8c708a0f86f4407f1f0b254cf Mon Sep 17 00:00:00 2001 From: XuPeng-SH Date: Fri, 4 Sep 2026 01:15:51 +0800 Subject: [PATCH 2/3] fix(ci): make stale-run cancellation generation-safe --- .github/workflows/static-checks.yml | 8 +- .github/workflows/supersede-pr-runs.yml | 24 ++- .github/workflows/test.yml | 10 +- CONTRIBUTING.md | 16 +- scripts/ci/cancel_stale_pr_runs.py | 162 ++++++++------- scripts/ci/test_cancel_stale_pr_runs.py | 254 +++++++++++++++++------- 6 files changed, 305 insertions(+), 169 deletions(-) diff --git a/.github/workflows/static-checks.yml b/.github/workflows/static-checks.yml index 3ff286308d..c6cb0b615b 100644 --- a/.github/workflows/static-checks.yml +++ b/.github/workflows/static-checks.yml @@ -41,7 +41,7 @@ jobs: check: needs: scope - if: ${{ always() }} + if: ${{ !cancelled() }} runs-on: ubuntu-latest timeout-minutes: 90 steps: @@ -50,7 +50,7 @@ jobs: persist-credentials: false - name: Validate repository metadata and documentation run: python3 scripts/ci/validate_repository.py - - name: Test CI scope routing + - name: Test CI contracts run: python3 -m unittest discover -s scripts/ci -p 'test_*.py' - name: Validate Terminal-Bench harness contracts if: ${{ needs.scope.result != 'success' || needs.scope.outputs.harness == 'true' }} @@ -95,7 +95,7 @@ jobs: sdk: name: "@astra/sdk (typecheck, test+coverage, build)" needs: scope - if: ${{ always() && (needs.scope.result != 'success' || needs.scope.outputs.sdk == 'true') }} + if: ${{ !cancelled() && (needs.scope.result != 'success' || needs.scope.outputs.sdk == 'true') }} runs-on: ubuntu-latest timeout-minutes: 45 steps: @@ -118,7 +118,7 @@ jobs: web: name: "web (typecheck, test, build)" needs: scope - if: ${{ always() && (needs.scope.result != 'success' || needs.scope.outputs.web == 'true') }} + if: ${{ !cancelled() && (needs.scope.result != 'success' || needs.scope.outputs.web == 'true') }} runs-on: ubuntu-latest timeout-minutes: 60 steps: diff --git a/.github/workflows/supersede-pr-runs.yml b/.github/workflows/supersede-pr-runs.yml index 7f7554c9da..1b94d67c43 100644 --- a/.github/workflows/supersede-pr-runs.yml +++ b/.github/workflows/supersede-pr-runs.yml @@ -5,36 +5,38 @@ on: branches: [main, develop] types: [synchronize] -# This controller runs from the trusted base revision. It must never check out +# This controller runs from its trusted workflow revision. It must never check out # or execute the pull request head while holding permission to cancel runs. permissions: actions: write contents: read - pull-requests: read -concurrency: - group: supersede-pr-runs-${{ github.event.pull_request.number }} - cancel-in-progress: true +# Do not add concurrency here. Every synchronize event owns one before -> after +# transition; dropping an intermediate controller could leave its prior head +# running after several rapid pushes. jobs: cancel-stale-runs: name: Cancel superseded PR runs + if: ${{ github.event.pull_request.head.repo.full_name != github.repository }} runs-on: ubuntu-latest timeout-minutes: 5 steps: - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7 with: - ref: ${{ github.event.pull_request.base.sha }} + ref: ${{ github.workflow_sha }} fetch-depth: 1 persist-credentials: false - name: Cancel runs for earlier PR heads env: GH_TOKEN: ${{ github.token }} - EVENT_HEAD_SHA: ${{ github.event.pull_request.head.sha }} - PR_NUMBER: ${{ github.event.pull_request.number }} + HEAD_REF: ${{ github.event.pull_request.head.ref }} + HEAD_REPOSITORY_ID: ${{ github.event.pull_request.head.repo.id }} REPOSITORY: ${{ github.repository }} + SUPERSEDED_HEAD_SHA: ${{ github.event.before }} run: >- python3 scripts/ci/cancel_stale_pr_runs.py - --repository "$REPOSITORY" - --pr-number "$PR_NUMBER" - --event-head-sha "$EVENT_HEAD_SHA" + --repository="$REPOSITORY" + --head-repository-id="$HEAD_REPOSITORY_ID" + --head-ref="$HEAD_REF" + --superseded-head-sha="$SUPERSEDED_HEAD_SHA" diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml index d71cc5bb8f..80b7746995 100644 --- a/.github/workflows/test.yml +++ b/.github/workflows/test.yml @@ -63,7 +63,7 @@ jobs: shard-a: name: "Test: astra-cli (${{ matrix.segment }})" needs: scope - if: ${{ always() }} + if: ${{ !cancelled() }} runs-on: ubuntu-latest timeout-minutes: 45 env: @@ -126,7 +126,7 @@ jobs: shard-b: name: "Test: astra-runtime" needs: scope - if: ${{ always() && (needs.scope.result != 'success' || needs.scope.outputs.test_runtime == 'true') }} + if: ${{ !cancelled() && (needs.scope.result != 'success' || needs.scope.outputs.test_runtime == 'true') }} runs-on: ubuntu-latest timeout-minutes: 35 steps: @@ -157,7 +157,7 @@ jobs: shard-c: name: "Test: turn-core + services + plan" needs: scope - if: ${{ always() && (needs.scope.result != 'success' || needs.scope.outputs.test_services == 'true') }} + if: ${{ !cancelled() && (needs.scope.result != 'success' || needs.scope.outputs.test_services == 'true') }} runs-on: ubuntu-latest timeout-minutes: 30 steps: @@ -177,7 +177,7 @@ jobs: shard-d: name: "Test: core crates + bridge hooks" needs: scope - if: ${{ always() && (needs.scope.result != 'success' || needs.scope.outputs.test_core == 'true') }} + if: ${{ !cancelled() && (needs.scope.result != 'success' || needs.scope.outputs.test_core == 'true') }} runs-on: ubuntu-latest timeout-minutes: 35 steps: @@ -209,7 +209,7 @@ jobs: test-online: name: "Test: online (${{ matrix.lane }})" needs: scope - if: ${{ always() }} + if: ${{ !cancelled() }} runs-on: ubuntu-latest timeout-minutes: 45 strategy: diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index a1c9360cb6..bfa88f1cce 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -89,11 +89,17 @@ contract and its tests live in [`scripts/ci/`](scripts/ci/). For fork pull requests, an update may require maintainer approval before the replacement test run can enter the normal concurrency group. A separate -base-revision controller cancels active runs for earlier heads immediately. The -controller never checks out or executes pull request code; the test workflows -remain low-privilege `pull_request` workflows. Normal concurrency groups use the -pull request number, so identically named branches in different forks remain -isolated. +trusted-workflow-revision controller cancels active runs for earlier heads +without waiting for replacement-run approval. The controller never checks out +or executes pull request code; the +test workflows remain low-privilege `pull_request` workflows. Normal concurrency +groups use the pull request number, so identically named branches in different +forks remain isolated. Each synchronize event cancels only its exact `before` +head and is intentionally not coalesced, so rapid pushes cannot make an older +controller cancel a newer generation or skip an intermediate cleanup. Jobs that +must run after scope-classification failure use `!cancelled()` rather than +`always()`, preserving fail-safe coverage without making superseded work resist +cancellation. ## Open a pull request diff --git a/scripts/ci/cancel_stale_pr_runs.py b/scripts/ci/cancel_stale_pr_runs.py index e94058c273..0dd27a2ba2 100755 --- a/scripts/ci/cancel_stale_pr_runs.py +++ b/scripts/ci/cancel_stale_pr_runs.py @@ -1,5 +1,5 @@ #!/usr/bin/env python3 -"""Cancel active GitHub Actions runs that belong to an earlier head of a PR.""" +"""Cancel active GitHub Actions runs for a superseded pull-request head.""" from __future__ import annotations @@ -12,32 +12,47 @@ from typing import Any, Protocol from urllib.error import HTTPError from urllib.parse import urlencode, urljoin, urlsplit -from urllib.request import Request, urlopen +from urllib.request import build_opener, HTTPRedirectHandler, Request -ACTIVE_RUN_STATUSES = ("queued", "in_progress", "pending", "requested", "waiting") -MAX_PAGES_PER_STATUS = 10 -MAX_STALE_RUNS = 1_000 +ACTIVE_RUN_STATUSES = frozenset( + {"queued", "in_progress", "pending", "requested", "waiting"} +) +# These caps keep the worst-case request path within the workflow's five-minute +# timeout while leaving headroom over Astra's three pull-request workflows. +MAX_PAGES = 5 +MAX_SUPERSEDED_RUNS = 10 +REQUEST_TIMEOUT_SECONDS = 10 REPOSITORY_PATTERN = re.compile(r"^[A-Za-z0-9_.-]+/[A-Za-z0-9_.-]+$") SHA_PATTERN = re.compile(r"^[0-9a-fA-F]{40}$") @dataclass(frozen=True) -class PullRequestHead: +class SupersededHead: repository_id: int - repository: str ref: str sha: str class ActionsApi(Protocol): - def get_pull_request(self, repository: str, number: int) -> dict[str, Any]: ... - - def list_runs(self, repository: str, status: str) -> list[dict[str, Any]]: ... + def list_runs(self, repository: str, head_sha: str) -> list[dict[str, Any]]: ... def cancel_run(self, repository: str, run_id: int) -> bool: ... +class RejectRedirects(HTTPRedirectHandler): + def redirect_request( + self, + req: Request, + fp: Any, + code: int, + msg: str, + headers: Any, + newurl: str, + ) -> Request | None: + raise RuntimeError("refusing to forward the GitHub token through a redirect") + + class GitHubApi: """Small GitHub REST client with bounded, origin-checked pagination.""" @@ -46,11 +61,17 @@ def __init__(self, token: str, api_url: str = "https://api.github.com") -> None: raise ValueError("GH_TOKEN is required") self._token = token self._api_url = api_url.rstrip("/") + "/" - self._api_origin = urlsplit(self._api_url).netloc + parsed_api_url = urlsplit(self._api_url) + if parsed_api_url.scheme.lower() != "https" or not parsed_api_url.netloc: + raise ValueError("GITHUB_API_URL must use HTTPS and include a host") + self._api_origin = (parsed_api_url.scheme.lower(), parsed_api_url.netloc.lower()) + self._opener = build_opener(RejectRedirects()) def _request(self, method: str, url: str) -> tuple[bytes, Any]: target = url if urlsplit(url).scheme else urljoin(self._api_url, url.lstrip("/")) - if urlsplit(target).netloc != self._api_origin: + parsed_target = urlsplit(target) + target_origin = (parsed_target.scheme.lower(), parsed_target.netloc.lower()) + if target_origin != self._api_origin: raise RuntimeError("refusing to send the GitHub token to another origin") request = Request( target, @@ -62,7 +83,7 @@ def _request(self, method: str, url: str) -> tuple[bytes, Any]: "X-GitHub-Api-Version": "2022-11-28", }, ) - with urlopen(request, timeout=30) as response: + with self._opener.open(request, timeout=REQUEST_TIMEOUT_SECONDS) as response: return response.read(), response.headers def _get_json(self, path: str) -> dict[str, Any]: @@ -80,17 +101,16 @@ def _next_link(headers: Any) -> str | None: return match.group(1) return None - def get_pull_request(self, repository: str, number: int) -> dict[str, Any]: - return self._get_json(f"repos/{repository}/pulls/{number}") - - def list_runs(self, repository: str, status: str) -> list[dict[str, Any]]: - query = urlencode({"event": "pull_request", "status": status, "per_page": 100}) + def list_runs(self, repository: str, head_sha: str) -> list[dict[str, Any]]: + query = urlencode( + {"event": "pull_request", "head_sha": head_sha, "per_page": 100} + ) next_url: str | None = f"repos/{repository}/actions/runs?{query}" runs: list[dict[str, Any]] = [] visited: set[str] = set() while next_url: - if next_url in visited or len(visited) >= MAX_PAGES_PER_STATUS: - raise RuntimeError(f"workflow-run pagination exceeded its safe bound for {status}") + if next_url in visited or len(visited) >= MAX_PAGES: + raise RuntimeError("workflow-run pagination exceeded its safe bound") visited.add(next_url) body, headers = self._request("GET", next_url) page = json.loads(body) @@ -106,102 +126,86 @@ def cancel_run(self, repository: str, run_id: int) -> bool: self._request("POST", f"repos/{repository}/actions/runs/{run_id}/cancel") except HTTPError as error: # A selected run can finish between the list and cancel requests. - if error.code == 409: + if error.code != 409: + raise + if error.fp is not None: + error.close() + run = self._get_json(f"repos/{repository}/actions/runs/{run_id}") + if run.get("status") == "completed": return False - raise + raise RuntimeError( + f"GitHub refused to cancel active workflow run {run_id}" + ) from error return True -def _pull_request_head(pull_request: dict[str, Any]) -> PullRequestHead: - head = pull_request.get("head") - if not isinstance(head, dict): - raise RuntimeError("pull request response omitted head") - repository = head.get("repo") - full_name = repository.get("full_name") if isinstance(repository, dict) else None - repository_id = repository.get("id") if isinstance(repository, dict) else None - ref = head.get("ref") - sha = head.get("sha") - if not isinstance(repository_id, int) or not all( - isinstance(value, str) and value for value in (full_name, ref, sha) - ): - raise RuntimeError("pull request response has an incomplete head identity") - return PullRequestHead(repository_id, full_name, ref, sha) - - -def select_stale_runs( - runs: list[dict[str, Any]], current_head: PullRequestHead +def select_superseded_runs( + runs: list[dict[str, Any]], superseded_head: SupersededHead ) -> list[dict[str, Any]]: - """Select prior-head runs for exactly the current fork repository and ref.""" + """Select active runs for exactly one superseded source-head generation.""" selected: dict[int, dict[str, Any]] = {} for run in runs: run_id = run.get("id") head_repository = run.get("head_repository") repository_id = head_repository.get("id") if isinstance(head_repository, dict) else None - if not isinstance(run_id, int): + if not isinstance(run_id, int) or run.get("status") not in ACTIVE_RUN_STATUSES: continue - if repository_id != current_head.repository_id: + if repository_id != superseded_head.repository_id: continue - if run.get("head_branch") != current_head.ref: + if run.get("head_branch") != superseded_head.ref: continue - if run.get("head_sha") == current_head.sha: + if run.get("head_sha") != superseded_head.sha: continue selected[run_id] = run return [selected[run_id] for run_id in sorted(selected)] -def cancel_stale_runs( - api: ActionsApi, repository: str, pr_number: int, event_head_sha: str +def cancel_superseded_runs( + api: ActionsApi, repository: str, superseded_head: SupersededHead ) -> tuple[int, int]: - pull_request = api.get_pull_request(repository, pr_number) - current_head = _pull_request_head(pull_request) - if pull_request.get("state") != "open" or current_head.sha != event_head_sha: - print( - "Ignoring a stale controller event: " - f"event head {event_head_sha}, live head {current_head.sha}, " - f"state {pull_request.get('state')}." - ) - return 0, 0 - - active_runs: list[dict[str, Any]] = [] - for status in ACTIVE_RUN_STATUSES: - active_runs.extend(api.list_runs(repository, status)) - stale_runs = select_stale_runs(active_runs, current_head) - if len(stale_runs) > MAX_STALE_RUNS: + runs = api.list_runs(repository, superseded_head.sha) + superseded_runs = select_superseded_runs(runs, superseded_head) + if len(superseded_runs) > MAX_SUPERSEDED_RUNS: raise RuntimeError( - f"refusing an unexpectedly large cancellation set ({len(stale_runs)} runs)" + "refusing an unexpectedly large cancellation set " + f"({len(superseded_runs)} runs)" ) cancelled = 0 - for run in stale_runs: + for run in superseded_runs: run_id = int(run["id"]) if api.cancel_run(repository, run_id): cancelled += 1 print( f"Cancellation requested for {run.get('name', 'workflow')} " - f"run {run_id} at {run.get('head_sha', '')}." + f"run {run_id} at {superseded_head.sha}." ) else: print(f"Run {run_id} became terminal before cancellation; no action needed.") print( - f"Found {len(stale_runs)} stale active run(s) for " - f"{current_head.repository}:{current_head.ref}; requested {cancelled} cancellation(s)." + f"Found {len(superseded_runs)} active run(s) for superseded head " + f"repository={superseded_head.repository_id}, ref={superseded_head.ref}, " + f"sha={superseded_head.sha}; requested {cancelled} cancellation(s)." ) - return len(stale_runs), cancelled + return len(superseded_runs), cancelled def parse_args(argv: list[str]) -> argparse.Namespace: parser = argparse.ArgumentParser() parser.add_argument("--repository", required=True) - parser.add_argument("--pr-number", required=True, type=int) - parser.add_argument("--event-head-sha", required=True) + parser.add_argument("--head-repository-id", required=True, type=int) + parser.add_argument("--head-ref", required=True) + parser.add_argument("--superseded-head-sha", required=True) args = parser.parse_args(argv) if not REPOSITORY_PATTERN.fullmatch(args.repository): parser.error("--repository must be an owner/repository pair") - if args.pr_number <= 0: - parser.error("--pr-number must be positive") - if not SHA_PATTERN.fullmatch(args.event_head_sha): - parser.error("--event-head-sha must be a 40-character hexadecimal commit SHA") + if args.head_repository_id <= 0: + parser.error("--head-repository-id must be positive") + if not args.head_ref: + parser.error("--head-ref must not be empty") + if not SHA_PATTERN.fullmatch(args.superseded_head_sha): + parser.error("--superseded-head-sha must be a 40-character hexadecimal SHA") return args @@ -211,7 +215,15 @@ def main(argv: list[str] | None = None) -> int: os.environ.get("GH_TOKEN", ""), os.environ.get("GITHUB_API_URL", "https://api.github.com"), ) - cancel_stale_runs(api, args.repository, args.pr_number, args.event_head_sha) + cancel_superseded_runs( + api, + args.repository, + SupersededHead( + repository_id=args.head_repository_id, + ref=args.head_ref, + sha=args.superseded_head_sha, + ), + ) return 0 diff --git a/scripts/ci/test_cancel_stale_pr_runs.py b/scripts/ci/test_cancel_stale_pr_runs.py index 3a24a4b517..08ee7da6c6 100755 --- a/scripts/ci/test_cancel_stale_pr_runs.py +++ b/scripts/ci/test_cancel_stale_pr_runs.py @@ -4,38 +4,46 @@ from contextlib import redirect_stdout from io import StringIO +from itertools import permutations import json from pathlib import Path import sys import unittest from unittest.mock import Mock +from urllib.error import HTTPError sys.path.insert(0, str(Path(__file__).resolve().parent)) from cancel_stale_pr_runs import ( # noqa: E402 ACTIVE_RUN_STATUSES, GitHubApi, - MAX_STALE_RUNS, - PullRequestHead, - cancel_stale_runs, - select_stale_runs, + MAX_SUPERSEDED_RUNS, + RejectRedirects, + SupersededHead, + cancel_superseded_runs, + parse_args, + select_superseded_runs, ) -CURRENT_SHA = "b" * 40 OLD_SHA = "a" * 40 +CURRENT_SHA = "b" * 40 +NEWER_SHA = "c" * 40 +LATEST_SHA = "d" * 40 def workflow_run( run_id: int, *, sha: str = OLD_SHA, + status: str = "in_progress", repository_id: int = 101, branch: str = "fix/runtime", ) -> dict[str, object]: return { "id": run_id, "name": "Test Suite", + "status": status, "head_sha": sha, "head_branch": branch, "head_repository": {"id": repository_id, "full_name": "contributor/Astra"}, @@ -43,27 +51,15 @@ def workflow_run( class FakeApi: - def __init__(self, *, live_sha: str = CURRENT_SHA, state: str = "open") -> None: - self.pull_request = { - "state": state, - "head": { - "sha": live_sha, - "ref": "fix/runtime", - "repo": {"id": 101, "full_name": "contributor/Astra"}, - }, - } - self.runs_by_status: dict[str, list[dict[str, object]]] = { - status: [] for status in ACTIVE_RUN_STATUSES - } + def __init__(self, runs: list[dict[str, object]]) -> None: + self.runs = runs + self.listed_heads: list[str] = [] self.cancelled: list[int] = [] self.finished_before_cancel: set[int] = set() - def get_pull_request(self, repository: str, number: int) -> dict[str, object]: - self.requested_pull = (repository, number) - return self.pull_request - - def list_runs(self, repository: str, status: str) -> list[dict[str, object]]: - return self.runs_by_status[status] + def list_runs(self, repository: str, head_sha: str) -> list[dict[str, object]]: + self.listed_heads.append(head_sha) + return self.runs def cancel_run(self, repository: str, run_id: int) -> bool: if run_id in self.finished_before_cancel: @@ -72,28 +68,53 @@ def cancel_run(self, repository: str, run_id: int) -> bool: return True -class SelectStaleRunsTests(unittest.TestCase): - def test_selects_only_prior_heads_of_the_exact_fork_ref(self) -> None: - head = PullRequestHead(101, "Contributor/Astra", "fix/runtime", CURRENT_SHA) +class SelectSupersededRunsTests(unittest.TestCase): + def test_selects_only_the_exact_fork_ref_sha_and_active_status(self) -> None: + head = SupersededHead(101, "fix/runtime", OLD_SHA) runs = [ workflow_run(1), workflow_run(2, sha=CURRENT_SHA), workflow_run(3, repository_id=202), workflow_run(4, branch="fix/other"), - {"id": 5, "head_sha": OLD_SHA, "head_branch": "fix/runtime"}, + workflow_run(5, status="completed"), + {"id": 6, "status": "queued", "head_sha": OLD_SHA}, ] - self.assertEqual([run["id"] for run in select_stale_runs(runs, head)], [1]) + self.assertEqual( + [run["id"] for run in select_superseded_runs(runs, head)], [1] + ) - def test_deduplicates_a_run_observed_during_status_transition(self) -> None: - head = PullRequestHead(101, "contributor/Astra", "fix/runtime", CURRENT_SHA) + def test_deduplicates_a_run_observed_during_pagination(self) -> None: + head = SupersededHead(101, "fix/runtime", OLD_SHA) run = workflow_run(7) - self.assertEqual([item["id"] for item in select_stale_runs([run, run], head)], [7]) + self.assertEqual( + [item["id"] for item in select_superseded_runs([run, run], head)], [7] + ) + + def test_every_controller_order_cancels_old_generations_only(self) -> None: + runs = [ + workflow_run(1, sha=OLD_SHA), + workflow_run(2, sha=CURRENT_SHA), + workflow_run(3, sha=NEWER_SHA), + workflow_run(4, sha=LATEST_SHA), + ] + + for controller_order in permutations((OLD_SHA, CURRENT_SHA, NEWER_SHA)): + with self.subTest(controller_order=controller_order): + selected = { + int(run["id"]) + for sha in controller_order + for run in select_superseded_runs( + runs, SupersededHead(101, "fix/runtime", sha) + ) + } + + self.assertEqual(selected, {1, 2, 3}) class GitHubApiTests(unittest.TestCase): - def test_list_runs_follows_pagination(self) -> None: + def test_list_runs_filters_by_superseded_sha_and_follows_pagination(self) -> None: api = GitHubApi("test-token") second_page = "https://api.github.com/repos/matrixorigin/Astra/actions/runs?page=2" api._request = Mock( # type: ignore[method-assign] @@ -107,73 +128,142 @@ def test_list_runs_follows_pagination(self) -> None: ) self.assertEqual( - [run["id"] for run in api.list_runs("matrixorigin/Astra", "in_progress")], + [run["id"] for run in api.list_runs("matrixorigin/Astra", OLD_SHA)], [1, 2], ) + first_url = api._request.call_args_list[0].args[1] + self.assertIn(f"head_sha={OLD_SHA}", first_url) + self.assertIn("event=pull_request", first_url) def test_request_rejects_a_foreign_origin_before_sending_the_token(self) -> None: api = GitHubApi("test-token") with self.assertRaisesRegex(RuntimeError, "another origin"): api._request("GET", "https://example.test/steal") + with self.assertRaisesRegex(RuntimeError, "another origin"): + api._request("GET", "http://api.github.com/cleartext") + + def test_client_rejects_a_cleartext_api_root(self) -> None: + with self.assertRaisesRegex(ValueError, "must use HTTPS"): + GitHubApi("test-token", "http://api.github.com") + + def test_redirect_handler_never_forwards_the_token(self) -> None: + handler = RejectRedirects() + + with self.assertRaisesRegex(RuntimeError, "through a redirect"): + handler.redirect_request( + Mock(), + Mock(), + 302, + "Found", + {}, + "https://example.test/steal", + ) + + def test_pagination_cycle_fails_instead_of_waiting_forever(self) -> None: + api = GitHubApi("test-token") + repeated_page = ( + "https://api.github.com/repos/matrixorigin/Astra/actions/runs?page=2" + ) + api._request = Mock( # type: ignore[method-assign] + return_value=( + json.dumps({"workflow_runs": []}).encode(), + {"Link": f'<{repeated_page}>; rel="next"'}, + ) + ) + with self.assertRaisesRegex(RuntimeError, "pagination"): + api.list_runs("matrixorigin/Astra", OLD_SHA) -class CancelStaleRunsTests(unittest.TestCase): - def test_cancels_prior_runs_across_every_active_status(self) -> None: - api = FakeApi() - for index, status in enumerate(ACTIVE_RUN_STATUSES, start=1): - api.runs_by_status[status] = [workflow_run(index)] - api.runs_by_status["pending"].append(workflow_run(100, sha=CURRENT_SHA)) + def test_cancel_treats_a_completion_race_as_terminal(self) -> None: + api = GitHubApi("test-token") + api._request = Mock( # type: ignore[method-assign] + side_effect=[ + HTTPError("url", 409, "Conflict", {}, None), + (json.dumps({"status": "completed"}).encode(), {}), + ] + ) - found, cancelled = cancel_stale_runs(api, "matrixorigin/Astra", 42, CURRENT_SHA) + self.assertFalse(api.cancel_run("matrixorigin/Astra", 42)) - self.assertEqual( - (found, cancelled), - (len(ACTIVE_RUN_STATUSES), len(ACTIVE_RUN_STATUSES)), + def test_cancel_does_not_hide_a_conflict_for_an_active_run(self) -> None: + api = GitHubApi("test-token") + api._request = Mock( # type: ignore[method-assign] + side_effect=[ + HTTPError("url", 409, "Conflict", {}, None), + (json.dumps({"status": "in_progress"}).encode(), {}), + ] ) - self.assertEqual(api.cancelled, list(range(1, len(ACTIVE_RUN_STATUSES) + 1))) - def test_stale_controller_cannot_cancel_a_newer_head(self) -> None: - api = FakeApi(live_sha="c" * 40) - api.runs_by_status["in_progress"] = [workflow_run(1, sha=CURRENT_SHA)] + with self.assertRaisesRegex(RuntimeError, "refused to cancel active"): + api.cancel_run("matrixorigin/Astra", 42) - found, cancelled = cancel_stale_runs(api, "matrixorigin/Astra", 42, CURRENT_SHA) - - self.assertEqual((found, cancelled), (0, 0)) - self.assertEqual(api.cancelled, []) - def test_closed_pull_request_event_is_a_safe_noop(self) -> None: - api = FakeApi(state="closed") - api.runs_by_status["in_progress"] = [workflow_run(1)] +class CancelSupersededRunsTests(unittest.TestCase): + def test_cancels_every_active_status_but_not_completed_runs(self) -> None: + runs = [ + workflow_run(index, status=status) + for index, status in enumerate(sorted(ACTIVE_RUN_STATUSES), start=1) + ] + runs.append(workflow_run(100, status="completed")) + api = FakeApi(runs) - found, cancelled = cancel_stale_runs(api, "matrixorigin/Astra", 42, CURRENT_SHA) + with redirect_stdout(StringIO()): + found, cancelled = cancel_superseded_runs( + api, + "matrixorigin/Astra", + SupersededHead(101, "fix/runtime", OLD_SHA), + ) - self.assertEqual((found, cancelled), (0, 0)) - self.assertEqual(api.cancelled, []) + self.assertEqual( + (found, cancelled), + (len(ACTIVE_RUN_STATUSES), len(ACTIVE_RUN_STATUSES)), + ) + self.assertEqual(api.listed_heads, [OLD_SHA]) def test_completion_race_is_not_reported_as_a_failed_cancellation(self) -> None: - api = FakeApi() - api.runs_by_status["in_progress"] = [workflow_run(1), workflow_run(2)] + api = FakeApi([workflow_run(1), workflow_run(2)]) api.finished_before_cancel.add(2) with redirect_stdout(StringIO()): - found, cancelled = cancel_stale_runs(api, "matrixorigin/Astra", 42, CURRENT_SHA) + found, cancelled = cancel_superseded_runs( + api, + "matrixorigin/Astra", + SupersededHead(101, "fix/runtime", OLD_SHA), + ) self.assertEqual((found, cancelled), (2, 1)) self.assertEqual(api.cancelled, [1]) def test_unexpected_cancellation_fanout_fails_before_mutation(self) -> None: - api = FakeApi() - api.runs_by_status["queued"] = [ - workflow_run(run_id) for run_id in range(1, MAX_STALE_RUNS + 2) - ] + api = FakeApi( + [workflow_run(run_id) for run_id in range(1, MAX_SUPERSEDED_RUNS + 2)] + ) with self.assertRaisesRegex(RuntimeError, "unexpectedly large"): - cancel_stale_runs(api, "matrixorigin/Astra", 42, CURRENT_SHA) + cancel_superseded_runs( + api, + "matrixorigin/Astra", + SupersededHead(101, "fix/runtime", OLD_SHA), + ) self.assertEqual(api.cancelled, []) +class ArgumentContractTests(unittest.TestCase): + def test_option_shaped_head_ref_remains_a_value(self) -> None: + args = parse_args( + [ + "--repository=matrixorigin/Astra", + "--head-repository-id=101", + "--head-ref=--repository", + f"--superseded-head-sha={OLD_SHA}", + ] + ) + + self.assertEqual(args.head_ref, "--repository") + + class WorkflowSecurityContractTests(unittest.TestCase): def test_controller_executes_only_the_trusted_base_revision(self) -> None: workflow = ( @@ -185,15 +275,26 @@ def test_controller_executes_only_the_trusted_base_revision(self) -> None: "pull_request_target:", "actions: write", "contents: read", - "pull-requests: read", - "ref: ${{ github.event.pull_request.base.sha }}", + "ref: ${{ github.workflow_sha }}", "persist-credentials: false", + "github.event.pull_request.head.repo.full_name != github.repository", + "SUPERSEDED_HEAD_SHA: ${{ github.event.before }}", "python3 scripts/ci/cancel_stale_pr_runs.py", + '--head-ref="$HEAD_REF"', ): with self.subTest(required=required): self.assertIn(required, workflow) - self.assertNotIn("ref: ${{ github.event.pull_request.head.sha }}", workflow) + permissions = workflow.partition("permissions:\n")[2].partition("\n\n")[0] + self.assertEqual(permissions, " actions: write\n contents: read") + self.assertEqual(workflow.count("uses: actions/checkout@"), 1) + refs = [ + line.strip() + for line in workflow.splitlines() + if line.strip().startswith("ref:") + ] + self.assertEqual(refs, ["ref: ${{ github.workflow_sha }}"]) self.assertNotIn("secrets.", workflow) + self.assertNotRegex(workflow, r"(?m)^\s*concurrency:") def test_pull_request_concurrency_isolated_by_pr_number(self) -> None: root = Path(__file__).resolve().parents[2] @@ -202,11 +303,26 @@ def test_pull_request_concurrency_isolated_by_pr_number(self) -> None: ".github/workflows/pr-title.yml", ".github/workflows/static-checks.yml", ".github/workflows/test.yml", - ".github/workflows/supersede-pr-runs.yml", ): with self.subTest(workflow=relative): self.assertIn(expected, (root / relative).read_text(encoding="utf-8")) + def test_pr_jobs_do_not_resist_workflow_cancellation(self) -> None: + root = Path(__file__).resolve().parents[2] + for relative in ( + ".github/workflows/static-checks.yml", + ".github/workflows/test.yml", + ): + workflow = (root / relative).read_text(encoding="utf-8") + job_level_always = [ + line + for line in workflow.splitlines() + if line.startswith(" if:") and "always()" in line + ] + with self.subTest(workflow=relative): + self.assertEqual(job_level_always, []) + self.assertIn("if: ${{ !cancelled()", workflow) + if __name__ == "__main__": unittest.main() From 4fe0a5ff3ba9430d275aa972f3c32748cd66e614 Mon Sep 17 00:00:00 2001 From: XuPeng-SH Date: Fri, 4 Sep 2026 01:25:26 +0800 Subject: [PATCH 3/3] fix(ci): harden stale-run cancellation failures --- scripts/ci/cancel_stale_pr_runs.py | 30 ++++++++++-- scripts/ci/test_cancel_stale_pr_runs.py | 62 +++++++++++++++++++++++-- 2 files changed, 84 insertions(+), 8 deletions(-) diff --git a/scripts/ci/cancel_stale_pr_runs.py b/scripts/ci/cancel_stale_pr_runs.py index 0dd27a2ba2..c605a11932 100755 --- a/scripts/ci/cancel_stale_pr_runs.py +++ b/scripts/ci/cancel_stale_pr_runs.py @@ -9,6 +9,7 @@ import os import re import sys +import time from typing import Any, Protocol from urllib.error import HTTPError from urllib.parse import urlencode, urljoin, urlsplit @@ -21,8 +22,10 @@ # These caps keep the worst-case request path within the workflow's five-minute # timeout while leaving headroom over Astra's three pull-request workflows. MAX_PAGES = 5 -MAX_SUPERSEDED_RUNS = 10 +MAX_SUPERSEDED_RUNS = 5 REQUEST_TIMEOUT_SECONDS = 10 +TERMINAL_STATUS_POLLS = 3 +TERMINAL_STATUS_POLL_SECONDS = 1 REPOSITORY_PATTERN = re.compile(r"^[A-Za-z0-9_.-]+/[A-Za-z0-9_.-]+$") SHA_PATTERN = re.compile(r"^[0-9a-fA-F]{40}$") @@ -130,9 +133,12 @@ def cancel_run(self, repository: str, run_id: int) -> bool: raise if error.fp is not None: error.close() - run = self._get_json(f"repos/{repository}/actions/runs/{run_id}") - if run.get("status") == "completed": - return False + for attempt in range(TERMINAL_STATUS_POLLS): + run = self._get_json(f"repos/{repository}/actions/runs/{run_id}") + if run.get("status") == "completed": + return False + if attempt + 1 < TERMINAL_STATUS_POLLS: + time.sleep(TERMINAL_STATUS_POLL_SECONDS) raise RuntimeError( f"GitHub refused to cancel active workflow run {run_id}" ) from error @@ -172,9 +178,18 @@ def cancel_superseded_runs( ) cancelled = 0 + failures: list[tuple[int, Exception]] = [] for run in superseded_runs: run_id = int(run["id"]) - if api.cancel_run(repository, run_id): + try: + cancellation_requested = api.cancel_run(repository, run_id) + except Exception as error: + failures.append((run_id, error)) + print( + f"Failed to cancel run {run_id}; continuing cleanup.", file=sys.stderr + ) + continue + if cancellation_requested: cancelled += 1 print( f"Cancellation requested for {run.get('name', 'workflow')} " @@ -188,6 +203,11 @@ def cancel_superseded_runs( f"repository={superseded_head.repository_id}, ref={superseded_head.ref}, " f"sha={superseded_head.sha}; requested {cancelled} cancellation(s)." ) + if failures: + failed_run_ids = ", ".join(str(run_id) for run_id, _ in failures) + raise RuntimeError( + f"failed to cancel {len(failures)} workflow run(s): {failed_run_ids}" + ) from failures[0][1] return len(superseded_runs), cancelled diff --git a/scripts/ci/test_cancel_stale_pr_runs.py b/scripts/ci/test_cancel_stale_pr_runs.py index 08ee7da6c6..ccfca39009 100755 --- a/scripts/ci/test_cancel_stale_pr_runs.py +++ b/scripts/ci/test_cancel_stale_pr_runs.py @@ -2,14 +2,14 @@ from __future__ import annotations -from contextlib import redirect_stdout +from contextlib import redirect_stderr, redirect_stdout from io import StringIO from itertools import permutations import json from pathlib import Path import sys import unittest -from unittest.mock import Mock +from unittest.mock import Mock, patch from urllib.error import HTTPError sys.path.insert(0, str(Path(__file__).resolve().parent)) @@ -192,11 +192,31 @@ def test_cancel_does_not_hide_a_conflict_for_an_active_run(self) -> None: side_effect=[ HTTPError("url", 409, "Conflict", {}, None), (json.dumps({"status": "in_progress"}).encode(), {}), + (json.dumps({"status": "in_progress"}).encode(), {}), + (json.dumps({"status": "in_progress"}).encode(), {}), ] ) - with self.assertRaisesRegex(RuntimeError, "refused to cancel active"): + with ( + patch("cancel_stale_pr_runs.time.sleep") as sleep, + self.assertRaisesRegex(RuntimeError, "refused to cancel active"), + ): api.cancel_run("matrixorigin/Astra", 42) + self.assertEqual(sleep.call_count, 2) + + def test_cancel_waits_for_an_already_requested_cancellation(self) -> None: + api = GitHubApi("test-token") + api._request = Mock( # type: ignore[method-assign] + side_effect=[ + HTTPError("url", 409, "Conflict", {}, None), + (json.dumps({"status": "in_progress"}).encode(), {}), + (json.dumps({"status": "completed"}).encode(), {}), + ] + ) + + with patch("cancel_stale_pr_runs.time.sleep") as sleep: + self.assertFalse(api.cancel_run("matrixorigin/Astra", 42)) + sleep.assert_called_once() class CancelSupersededRunsTests(unittest.TestCase): @@ -235,6 +255,27 @@ def test_completion_race_is_not_reported_as_a_failed_cancellation(self) -> None: self.assertEqual((found, cancelled), (2, 1)) self.assertEqual(api.cancelled, [1]) + def test_one_failure_does_not_prevent_other_runs_from_being_cancelled(self) -> None: + api = Mock() + api.list_runs.return_value = [workflow_run(1), workflow_run(2)] + api.cancel_run.side_effect = [RuntimeError("transient failure"), True] + + with ( + redirect_stderr(StringIO()), + redirect_stdout(StringIO()), + self.assertRaisesRegex(RuntimeError, "failed to cancel 1 workflow run"), + ): + cancel_superseded_runs( + api, + "matrixorigin/Astra", + SupersededHead(101, "fix/runtime", OLD_SHA), + ) + + self.assertEqual( + [call.args for call in api.cancel_run.call_args_list], + [("matrixorigin/Astra", 1), ("matrixorigin/Astra", 2)], + ) + def test_unexpected_cancellation_fanout_fails_before_mutation(self) -> None: api = FakeApi( [workflow_run(run_id) for run_id in range(1, MAX_SUPERSEDED_RUNS + 2)] @@ -307,6 +348,21 @@ def test_pull_request_concurrency_isolated_by_pr_number(self) -> None: with self.subTest(workflow=relative): self.assertIn(expected, (root / relative).read_text(encoding="utf-8")) + def test_cancellation_bound_covers_every_pull_request_workflow(self) -> None: + workflows = Path(__file__).resolve().parents[2] / ".github/workflows" + pull_request_workflows = [ + path.name + for pattern in ("*.yml", "*.yaml") + for path in workflows.glob(pattern) + if "\n pull_request:\n" in path.read_text(encoding="utf-8") + ] + + self.assertEqual( + sorted(pull_request_workflows), + ["pr-title.yml", "static-checks.yml", "test.yml"], + ) + self.assertLessEqual(len(pull_request_workflows), MAX_SUPERSEDED_RUNS) + def test_pr_jobs_do_not_resist_workflow_cancellation(self) -> None: root = Path(__file__).resolve().parents[2] for relative in (