From 0d1b312673d062e7ca236d2f444728a6afcfed5e Mon Sep 17 00:00:00 2001 From: Zhengchao An Date: Sat, 5 Sep 2026 16:49:58 +0800 Subject: [PATCH] fix(ci): require fresh successful scheduled validations (#7192) --- .../scheduled-validation-freshness.yml | 1 + .../check_scheduled_validation_freshness.py | 398 +++++++++++++----- 2 files changed, 293 insertions(+), 106 deletions(-) diff --git a/.github/workflows/scheduled-validation-freshness.yml b/.github/workflows/scheduled-validation-freshness.yml index eef340869..69ddc844c 100644 --- a/.github/workflows/scheduled-validation-freshness.yml +++ b/.github/workflows/scheduled-validation-freshness.yml @@ -42,6 +42,7 @@ jobs: - name: Check latest scheduled runs env: GH_TOKEN: ${{ secrets.GITHUB_TOKEN }} + RUSTFS_DEFAULT_BRANCH: ${{ github.event.repository.default_branch }} run: | set +e python3 scripts/check_scheduled_validation_freshness.py \ diff --git a/scripts/check_scheduled_validation_freshness.py b/scripts/check_scheduled_validation_freshness.py index a5812c7a3..c694e3c89 100644 --- a/scripts/check_scheduled_validation_freshness.py +++ b/scripts/check_scheduled_validation_freshness.py @@ -1,10 +1,11 @@ #!/usr/bin/env python3 -"""Fail when a critical scheduled validation has not started recently.""" +"""Require recent scheduled attempts and completed successes on the default branch.""" from __future__ import annotations import argparse from datetime import datetime, timedelta, timezone +import io import json import os from pathlib import Path @@ -13,7 +14,7 @@ import sys import tempfile import unittest from unittest import mock -from urllib.parse import quote, urlencode +from urllib.parse import parse_qs, quote, urlencode, urlsplit from urllib.request import Request, urlopen @@ -75,15 +76,8 @@ def stale_reason( run: dict[str, object] | None, now: datetime, max_age_hours: int, - never_ran_grace_until: datetime | None = None, ) -> str | None: if run is None: - # The grace deadline only covers a workflow whose first scheduled slot - # has not arrived yet (for example a monthly cron enabled mid-month). - # A recorded-but-old run proves the schedule used to fire and stopped, - # so the grace never masks that case. - if never_ran_grace_until is not None and now <= never_ran_grace_until: - return None return "no scheduled run has been recorded" created_at = parse_timestamp(run.get("created_at")) age = now - created_at @@ -93,14 +87,23 @@ def stale_reason( def fetch_latest_scheduled_run( - repository: str, workflow: str, token: str, api_url: str + repository: str, + workflow: str, + token: str, + api_url: str, + default_branch: str, + successful: bool = False, ) -> dict[str, object] | None: owner, repo = repository.split("/", 1) workflow_name = Path(workflow).name + query = {"event": "schedule", "branch": default_branch, "per_page": 1} + if successful: + # Filter on the server: the last success may be beyond a page of failures. + query["status"] = "success" endpoint = ( f"{api_url.rstrip('/')}/repos/{quote(owner, safe='')}/{quote(repo, safe='')}" f"/actions/workflows/{quote(workflow_name, safe='')}/runs?" - + urlencode({"event": "schedule", "per_page": 1}) + + urlencode(query) ) request = Request( endpoint, @@ -110,57 +113,104 @@ def fetch_latest_scheduled_run( "X-GitHub-Api-Version": "2022-11-28", }, ) - with urlopen(request, timeout=30) as response: + # Two requests per manifest entry must fit the watchdog's ten-minute job. + with urlopen(request, timeout=15) as response: payload = json.load(response) - runs = payload.get("workflow_runs") + runs = payload.get("workflow_runs") if isinstance(payload, dict) else None if not isinstance(runs, list): raise ValueError(f"GitHub returned no workflow_runs list for {workflow}") + total_count = payload.get("total_count") + if not isinstance(total_count, int) or isinstance(total_count, bool) or total_count < len(runs): + raise ValueError(f"GitHub returned an invalid run count for {workflow}") if not runs: + if total_count: + raise ValueError(f"GitHub returned an empty first page with recorded runs for {workflow}") return None - if not isinstance(runs[0], dict): + run = runs[0] + if not isinstance(run, dict): raise ValueError(f"GitHub returned an invalid workflow run for {workflow}") - return runs[0] + if run.get("event") != "schedule" or run.get("head_branch") != default_branch: + raise ValueError(f"GitHub returned a run outside the scheduled default-branch query for {workflow}") + if not isinstance(run.get("status"), str) or not run["status"]: + raise ValueError(f"GitHub returned no run status for {workflow}") + conclusion = run.get("conclusion") + if (conclusion is not None and not isinstance(conclusion, str)) or ( + run["status"] == "completed" and not conclusion + ): + raise ValueError(f"GitHub returned an invalid run conclusion for {workflow}") + if successful and (run["status"] != "completed" or conclusion != "success"): + raise ValueError(f"GitHub returned a run without a completed success for {workflow}") + parse_timestamp(run.get("created_at")) + if not isinstance(run.get("html_url"), str) or not run["html_url"]: + raise ValueError(f"GitHub returned no run URL for {workflow}") + return run -def write_report(path: Path, failures: list[tuple[str, int, str, str]]) -> None: - lines = ["## Scheduled validation freshness"] - if not failures: - lines.append("") - lines.append("All critical scheduled validations have a recent scheduled run.") - else: - lines.extend( - [ - "", - "The following critical validations are stale or could not be inspected:", - "", - "| Workflow | Limit | Result | Last run |", - "| --- | ---: | --- | --- |", - ] - ) - for workflow, max_age_hours, reason, run_url in failures: - link = f"[open]({run_url})" if run_url else "—" - lines.append(f"| `{workflow}` | {max_age_hours}h | {reason} | {link} |") +def describe_run(run: dict[str, object] | None) -> str: + if run is None: + return "No recorded run" + outcome = run["status"] + if run.get("conclusion"): + outcome = f"{outcome}/{run['conclusion']}" + return f"[{outcome}]({run['html_url']}) — created {run['created_at']}" + + +def write_report(path: Path, rows: list[tuple[str, int, str, str, str]], default_branch: str) -> None: + lines = [ + "## Scheduled validation freshness", + "", + f"Default branch: `{default_branch}`. Ages use scheduled-run creation time; rerunning an old commit does not refresh its evidence.", + "Attempt outcomes are shown independently of successful-run freshness.", + "Success is the GitHub workflow run conclusion; suite completeness remains the responsibility of each workflow.", + "", + "| Workflow | Limit | Freshness | Last attempt | Last completed success |", + "| --- | ---: | --- | --- | --- |", + ] + for workflow, max_age_hours, result, attempt, success in rows: + cells = [f"`{workflow}`", f"{max_age_hours}h", result, attempt, success] + lines.append("| " + " | ".join(cell.replace("|", "\\|").replace("\n", " ") for cell in cells) + " |") path.write_text("\n".join(lines) + "\n") def check_freshness( - config: Path, report: Path, repository: str, token: str, api_url: str + config: Path, report: Path, repository: str, token: str, api_url: str, default_branch: str ) -> int: now = datetime.now(timezone.utc) - failures: list[tuple[str, int, str, str]] = [] + rows: list[tuple[str, int, str, str, str]] = [] + failed = False for workflow, max_age_hours, never_ran_grace_until in load_validations(config): - try: - run = fetch_latest_scheduled_run(repository, workflow, token, api_url) - reason = stale_reason(run, now, max_age_hours, never_ran_grace_until) - if reason is not None: - run_url = str(run.get("html_url", "")) if run else "" - failures.append((workflow, max_age_hours, reason, run_url)) - except Exception as error: - failures.append( - (workflow, max_age_hours, f"inspection failed: {error}", "") - ) - write_report(report, failures) - return 1 if failures else 0 + runs: dict[str, dict[str, object] | None] = {} + reasons: list[str] = [] + for label, successful in (("Last attempt", False), ("Last completed success", True)): + try: + runs[label] = fetch_latest_scheduled_run( + repository, workflow, token, api_url, default_branch, successful + ) + except Exception as error: + reasons.append(f"{label}: inspection failed: {error}") + # A failed inspection or any recorded attempt ends first-run grace. + initial_grace = ( + len(runs) == 2 + and all(run is None for run in runs.values()) + and never_ran_grace_until is not None + and now <= never_ran_grace_until + ) + if not initial_grace: + for label, run in runs.items(): + reason = stale_reason(run, now, max_age_hours) + if reason is not None: + reasons.append(f"{label}: {reason}") + failed |= bool(reasons) + result = "; ".join(reasons) if reasons else "Fresh" + if initial_grace: + result = f"Initial grace until {never_ran_grace_until.isoformat()}" + evidence = [ + describe_run(runs[label]) if label in runs else "Inspection failed" + for label in ("Last attempt", "Last completed success") + ] + rows.append((workflow, max_age_hours, result, *evidence)) + write_report(report, rows, default_branch) + return 1 if failed else 0 class SelfTests(unittest.TestCase): @@ -173,15 +223,6 @@ class SelfTests(unittest.TestCase): self.assertIsNotNone(stale_reason(past_limit, self.NOW, 36)) self.assertIsNotNone(stale_reason(None, self.NOW, 36)) - def test_never_ran_grace_only_covers_missing_runs(self) -> None: - future_grace = self.NOW + timedelta(hours=1) - past_grace = self.NOW - timedelta(seconds=1) - self.assertIsNone(stale_reason(None, self.NOW, 36, future_grace)) - self.assertIsNone(stale_reason(None, self.NOW, 36, self.NOW)) - self.assertIsNotNone(stale_reason(None, self.NOW, 36, past_grace)) - stale_run = {"created_at": "2026-08-20T23:59:59Z"} - self.assertIsNotNone(stale_reason(stale_run, self.NOW, 36, future_grace)) - def test_config_rejects_duplicate_and_invalid_entries(self) -> None: with tempfile.TemporaryDirectory() as tmp: path = Path(tmp) / "validations.json" @@ -238,63 +279,205 @@ class SelfTests(unittest.TestCase): ], ) - def test_check_reports_missing_runs(self) -> None: + @staticmethod + def run_fixture(**overrides: object) -> dict[str, object]: + return { + "status": "completed", + "conclusion": "success", + "event": "schedule", + "head_branch": "release/current", + "created_at": "2026-08-22T00:00:00Z", + "html_url": "https://github.test/rustfs/rustfs/actions/runs/1", + **overrides, + } + + def check_payloads( + self, payloads: list[object], *, grace: str | None = None, workflows: int = 1 + ) -> tuple[int, str, list]: with tempfile.TemporaryDirectory() as tmp: root = Path(tmp) config = root / "validations.json" report = root / "report.md" - config.write_text( - json.dumps( - [ - {"workflow": ".github/workflows/ci.yml", "max_age_hours": 36}, - {"workflow": ".github/workflows/fuzz.yml", "max_age_hours": 36}, - {"workflow": ".github/workflows/mint.yml", "max_age_hours": 36}, - ] - ) - ) - with mock.patch( - __name__ + ".fetch_latest_scheduled_run", - side_effect=[ - {"created_at": "2999-01-01T00:00:00Z"}, - None, - RuntimeError("API unavailable"), - ], + entries = [ + {"workflow": f".github/workflows/check-{index}.yml", "max_age_hours": 36} + for index in range(workflows) + ] + if grace is not None: + entries[0]["never_ran_grace_until"] = grace + config.write_text(json.dumps(entries)) + responses = [] + for payload in payloads: + if isinstance(payload, dict) and isinstance(payload.get("workflow_runs"), list): + payload = {"total_count": len(payload["workflow_runs"]), **payload} + responses.append(payload if isinstance(payload, Exception) else io.StringIO(json.dumps(payload))) + with ( + mock.patch(__name__ + ".urlopen", side_effect=responses) as request, + mock.patch(__name__ + ".datetime", wraps=datetime) as clock, ): - self.assertEqual( - check_freshness( - config, - report, - "rustfs/rustfs", - "token", - "https://api.github.test", - ), - 1, + clock.now.return_value = self.NOW + status = check_freshness( + config, report, "rustfs/rustfs", "test-token", + "https://api.github.test", "release/current", ) - contents = report.read_text() - self.assertIn(".github/workflows/fuzz.yml", contents) - self.assertIn("inspection failed: API unavailable", contents) - self.assertNotIn(".github/workflows/ci.yml`", contents) + return status, report.read_text(), request.call_args_list - config.write_text( - json.dumps( - [{"workflow": ".github/workflows/ci.yml", "max_age_hours": 36}] + def test_requests_filter_schedule_default_branch_and_success_on_server(self) -> None: + attempt = self.run_fixture(status="in_progress", conclusion=None) + success = self.run_fixture(html_url="https://github.test/rustfs/rustfs/actions/runs/2") + status, report, calls = self.check_payloads([ + {"workflow_runs": [attempt], "total_count": 1001}, + {"workflow_runs": [success], "total_count": 1}, + ]) + self.assertEqual(status, 0) + self.assertEqual(len(calls), 2) + for call, successful in zip(calls, (False, True)): + request = call.args[0] + url = urlsplit(request.full_url) + self.assertEqual(url.path, "/repos/rustfs/rustfs/actions/workflows/check-0.yml/runs") + expected = {"event": ["schedule"], "branch": ["release/current"], "per_page": ["1"]} + if successful: + expected["status"] = ["success"] + self.assertEqual(parse_qs(url.query), expected) + self.assertEqual(request.get_header("Authorization"), "Bearer test-token") + self.assertEqual(call.kwargs, {"timeout": 15}) + self.assertIn("[in_progress]", report) + self.assertIn(str(attempt["html_url"]), report) + self.assertIn(str(success["html_url"]), report) + + def test_cancelled_attempt_cannot_refresh_expired_success(self) -> None: + attempt = self.run_fixture(conclusion="cancelled") + success = self.run_fixture( + created_at="2026-08-20T23:59:59Z", updated_at="2026-08-22T11:59:59Z", + html_url="https://github.test/rustfs/rustfs/actions/runs/2", + ) + status, report, _ = self.check_payloads([ + {"workflow_runs": [attempt]}, {"workflow_runs": [success]}, + ]) + self.assertEqual(status, 1) + self.assertIn("Last completed success: last scheduled run is", report) + self.assertIn("[completed/cancelled]", report) + for run in (attempt, success): + self.assertIn(str(run["html_url"]), report) + self.assertIn(str(run["created_at"]), report) + + def test_attempt_outcome_does_not_replace_recent_success(self) -> None: + success = self.run_fixture(created_at="2026-08-21T00:00:00Z") + for state, conclusion in ( + ("completed", "failure"), ("completed", "cancelled"), + ("completed", "timed_out"), ("completed", "success"), + ("queued", None), ("in_progress", None), + ): + with self.subTest(state=state, conclusion=conclusion): + status, report, _ = self.check_payloads([ + {"workflow_runs": [self.run_fixture(status=state, conclusion=conclusion)]}, + {"workflow_runs": [success]}, + ]) + self.assertEqual(status, 0) + self.assertIn(f"[{state}" + (f"/{conclusion}" if conclusion else "") + "]", report) + self.assertIn("Fresh", report) + self.assertNotIn("All critical scheduled validations", report) + + def test_grace_requires_two_successful_queries_with_no_history(self) -> None: + for attempt, success, grace, expected in ( + (None, None, "2026-08-22T12:00:00Z", 0), + (None, None, "2026-08-22T11:59:59Z", 1), + (self.run_fixture(conclusion="failure"), None, "2026-08-23T00:00:00Z", 1), + (self.run_fixture(status="queued", conclusion=None), None, "2026-08-23T00:00:00Z", 1), + (None, self.run_fixture(), "2026-08-23T00:00:00Z", 1), + ): + with self.subTest(attempt=attempt, success=success, grace=grace): + status, report, _ = self.check_payloads([ + {"workflow_runs": [] if attempt is None else [attempt]}, + {"workflow_runs": [] if success is None else [success]}, + ], grace=grace) + self.assertEqual(status, expected) + self.assertEqual("Initial grace until" in report, expected == 0) + + def test_api_failures_preserve_other_evidence_and_never_enter_grace(self) -> None: + good = {"workflow_runs": [self.run_fixture()]} + for first, second in ( + (RuntimeError("API unavailable"), good), + (good, RuntimeError("API unavailable")), + (RuntimeError("API unavailable"), {"workflow_runs": []}), + ): + with self.subTest(first=first, second=second): + status, report, calls = self.check_payloads( + [first, second], grace="2026-08-23T00:00:00Z" ) - ) - with mock.patch( - __name__ + ".fetch_latest_scheduled_run", - return_value={"created_at": "2999-01-01T00:00:00Z"}, - ): - self.assertEqual( - check_freshness( - config, - report, - "rustfs/rustfs", - "token", - "https://api.github.test", - ), - 0, - ) - self.assertIn("All critical scheduled validations", report.read_text()) + self.assertEqual(status, 1) + self.assertEqual(len(calls), 2) + self.assertIn("inspection failed: API unavailable", report) + self.assertNotIn("Initial grace until", report) + if first is good or second is good: + self.assertIn(str(self.run_fixture()["html_url"]), report) + + def test_invalid_api_evidence_fails_closed(self) -> None: + malformed = [ + [], {}, {"workflow_runs": {}}, {"workflow_runs": [None]}, + {"workflow_runs": [], "total_count": 1}, + {"workflow_runs": [], "total_count": -1}, + {"workflow_runs": [], "total_count": None}, + {"workflow_runs": [], "total_count": True}, + *({"workflow_runs": [self.run_fixture(**override)]} for override in ( + {"event": "workflow_dispatch"}, {"head_branch": "other"}, + {"created_at": "invalid"}, {"created_at": "2026-08-22T00:00:00"}, + {"status": None}, {"conclusion": None}, {"conclusion": 1}, + {"html_url": ""}, + )), + ] + for payload in malformed: + for index, label in enumerate(("Last attempt", "Last completed success")): + with self.subTest(payload=payload, label=label): + payloads = [{"workflow_runs": [self.run_fixture()]} for _ in range(2)] + payloads[index] = payload + status, report, _ = self.check_payloads(payloads, grace="2026-08-23T00:00:00Z") + self.assertEqual(status, 1) + self.assertIn(f"{label}: inspection failed", report) + self.assertNotIn("Initial grace until", report) + self.assertIn(str(self.run_fixture()["html_url"]), report) + for state, conclusion in (("in_progress", "success"), ("completed", "failure"), ("completed", "skipped")): + with self.subTest(state=state, conclusion=conclusion): + status, report, _ = self.check_payloads([ + {"workflow_runs": [self.run_fixture()]}, + {"workflow_runs": [self.run_fixture(status=state, conclusion=conclusion)]}, + ]) + self.assertEqual(status, 1) + self.assertIn("without a completed success", report) + + def test_report_retains_every_workflow(self) -> None: + status, report, calls = self.check_payloads([ + {"workflow_runs": [self.run_fixture()]}, {"workflow_runs": [self.run_fixture()]}, + {"workflow_runs": []}, {"workflow_runs": []}, + RuntimeError("API unavailable"), {"workflow_runs": [self.run_fixture()]}, + ], workflows=3) + self.assertEqual(status, 1) + self.assertEqual(len(calls), 6) + for index in range(3): + self.assertEqual(report.count(f"`.github/workflows/check-{index}.yml`"), 1) + self.assertIn("No recorded run", report) + self.assertIn("Inspection failed", report) + + def test_cli_requires_the_repository_default_branch(self) -> None: + from check_test_wiring import yaml_block + + workflow = (ROOT / ".github/workflows/scheduled-validation-freshness.yml").read_text().splitlines() + job = yaml_block(workflow, "check-freshness", 2) + self.assertIsNotNone(job) + start = job.index(" - name: Check latest scheduled runs") + end = next((index for index in range(start + 1, len(job)) if job[index].startswith(" - ")), len(job)) + environment = yaml_block(job[start:end], "env", 8) + self.assertIsNotNone(environment) + self.assertIn(" RUSTFS_DEFAULT_BRANCH: ${{ github.event.repository.default_branch }}", environment) + + with ( + mock.patch.dict(os.environ, {"GITHUB_REPOSITORY": "rustfs/rustfs", "GH_TOKEN": "test-token"}, clear=True), + mock.patch.object(sys, "argv", ["checker", "--report", "unused.md"]), + mock.patch("sys.stderr", new=io.StringIO()) as stderr, + self.assertRaises(SystemExit) as error, + ): + main() + self.assertEqual(error.exception.code, 2) + self.assertIn("RUSTFS_DEFAULT_BRANCH", stderr.getvalue()) def main() -> int: @@ -318,11 +501,14 @@ def main() -> int: repository = os.environ.get("GITHUB_REPOSITORY", "") token = os.environ.get("GH_TOKEN", "") api_url = os.environ.get("GITHUB_API_URL", "https://api.github.com") + default_branch = os.environ.get("RUSTFS_DEFAULT_BRANCH", "") if not re.fullmatch(r"[^/\s]+/[^/\s]+", repository): parser.error("GITHUB_REPOSITORY must be owner/repository") if not token: parser.error("GH_TOKEN is required") - return check_freshness(args.config, args.report, repository, token, api_url) + if not default_branch or any(character.isspace() for character in default_branch): + parser.error("RUSTFS_DEFAULT_BRANCH is required and must name the repository default branch") + return check_freshness(args.config, args.report, repository, token, api_url, default_branch) if __name__ == "__main__":