ci: reduce repeated PR runner work (#8463)

scripts/ci_gate.py

@@ -21,6 +21,9 @@ CODE_JOBS = (
     "build-rustfs-debug-binary", "uring-integration", "e2e-tests",
     "s3-implemented-tests", "s3-lifecycle-behavior-tests",
 )
+# Full alternate-feature tests remain mandatory on main, merge groups, and
+# scheduled/manual runs, without repeating four builds on every PR revision.
+EXTENDED_FEATURE_JOBS = ("test-and-lint-rio-v2", "test-and-lint-protocols")
 # These jobs still exercise the server and its black-box harness when only E2E
 # Rust sources change. Production code, manifests and shared test configuration
 # always select the full matrix. No production package depends on e2e_test.
@@ -127,6 +130,8 @@ def expected_results(mode: str, event: str, ref: str) -> dict[str, str]:
         raise ValueError(f"invalid CI selection: {mode!r} for {event!r}")
     expected = {job: "success" for job in ALWAYS_JOBS}
     expected.update({job: "success" if mode == "full" or (mode == "e2e" and job in E2E_JOBS) else "skipped" for job in CODE_JOBS})
+    if event == "pull_request":
+        expected.update({job: "skipped" for job in EXTENDED_FEATURE_JOBS})
     rio = mode == "full" and event in ("schedule", "workflow_dispatch")
     expected.update({job: "success" if rio else "skipped" for job in OPTIONAL_JOBS[:2]})
     full = mode == "full" and (event in ("merge_group", "workflow_dispatch") or (event == "push" and ref == "refs/heads/main"))
@@ -215,6 +220,10 @@ def check_workflow(root: Path) -> list[str]:
         errors.append("Test and Lint must uniquely name the aggregate gate")
     if scalar(gate, "if", 4) != "always() && (github.event_name != 'pull_request' || github.event.action != 'closed')":
         errors.append("required-checks must run after failed or skipped dependencies")
+    for job in EXTENDED_FEATURE_JOBS:
+        block = yaml_block(jobs, job, 2) or []
+        if scalar(block, "if", 4) != "needs.classify-changes.outputs.mode == 'full' && github.event_name != 'pull_request'":
+            errors.append(f"{job} must retain full validation on every non-PR event")
     if scalar(gate, "shell", 8) != "bash" or scalar(gate, "run", 8) != "python3 scripts/ci_gate.py verify" or scalar(gate, "CI_NEEDS", 10) != "${{ toJSON(needs) }}":
         errors.append("required-checks must verify the actual needs results")
     if any(re.match(r'''\s+(?:- )?(?:["']?continue-on-error["']?\s*:|["']?if["']?\s*:)''', line) and not line.startswith("    if:") for line in gate):
@@ -315,13 +324,14 @@ class SelfTests(unittest.TestCase):
 
     def test_event_contract_requires_complete_candidate_and_optional_lanes(self):
         ordinary = expected_results("full", "pull_request", "refs/pull/1/merge")
-        self.assertEqual({job for job, state in ordinary.items() if state == "skipped"}, set(OPTIONAL_JOBS))
+        self.assertEqual({job for job, state in ordinary.items() if state == "skipped"}, set(OPTIONAL_JOBS + EXTENDED_FEATURE_JOBS))
         docs = expected_results("docs", "pull_request", "refs/pull/1/merge")
         self.assertEqual({job for job, state in docs.items() if state == "success"}, set(ALWAYS_JOBS))
         e2e = expected_results("e2e", "pull_request", "refs/pull/1/merge")
         self.assertEqual({job for job, state in e2e.items() if state == "success"}, set(ALWAYS_JOBS + E2E_JOBS))
         for event in ("schedule", "workflow_dispatch", "merge_group", "push"):
             result = expected_results("full", event, "refs/heads/main")
+            self.assertTrue(all(result[job] == "success" for job in CODE_JOBS))
             self.assertEqual(result["e2e-full"], "skipped" if event == "schedule" else "success")
             self.assertEqual(result["e2e-tests-rio-v2"], "success" if event in ("schedule", "workflow_dispatch") else "skipped")
             with self.assertRaises(ValueError):
@@ -330,7 +340,7 @@ class SelfTests(unittest.TestCase):
                 expected_results("e2e", event, "refs/heads/main")
 
     def test_every_wrong_result_missing_job_or_selection_fails_closed(self):
-        for mode, event in (("full", "pull_request"), ("docs", "pull_request"), ("e2e", "pull_request"), ("full", "schedule"), ("full", "workflow_dispatch"), ("full", "merge_group")):
+        for mode, event in (("full", "pull_request"), ("docs", "pull_request"), ("e2e", "pull_request"), ("full", "push"), ("full", "schedule"), ("full", "workflow_dispatch"), ("full", "merge_group")):
             good = {job: {"result": value} for job, value in expected_results(mode, event, "refs/heads/main").items()}
             good["classify-changes"]["outputs"] = {"mode": mode}
             self.assertEqual(verify_results(good, event, "refs/heads/main"), [])
@@ -377,6 +387,12 @@ class SelfTests(unittest.TestCase):
             (root / ".github/workflows").mkdir(parents=True)
             source = (ROOT / ".github/workflows/ci.yml").read_text()
             path = root / ".github/workflows/ci.yml"
+            for job in EXTENDED_FEATURE_JOBS:
+                before, block = source.split(f"  {job}:\n", 1)
+                for condition in ("false", "github.event_name == 'schedule'", "github.event_name != 'pull_request'"):
+                    changed = re.sub(r"^    if: .*", f"    if: {condition}", block, count=1, flags=re.MULTILINE)
+                    path.write_text(before + f"  {job}:\n" + changed)
+                    self.assertIn(f"{job} must retain full validation on every non-PR event", check_workflow(root))
             for job in ALWAYS_JOBS + CODE_JOBS + OPTIONAL_JOBS:
                 before, gate = source.split("  required-checks:\n", 1)
                 path.write_text(before + "  required-checks:\n" + gate.replace(f"      - {job}\n", "", 1))

scripts/test_upgrade_workflow.py

@@ -0,0 +1,99 @@
+#!/usr/bin/env python3
+"""Exercise the consolidated upgrade lane without building RustFS."""
+
+from __future__ import annotations
+
+import json
+import os
+from pathlib import Path
+import re
+import subprocess
+import sys
+import tempfile
+import unittest
+
+from check_test_wiring import yaml_block
+from test_security_workflow import named_steps, shell_body
+
+
+ROOT = Path(__file__).resolve().parents[1]
+
+
+class UpgradeWorkflowTests(unittest.TestCase):
+    def setUp(self):
+        source = (ROOT / ".github/workflows/e2e-upgrade.yml").read_text().splitlines()
+        self.job = yaml_block(source, "upgrade", 2)
+        self.steps = named_steps(self.job)
+        self.step = self.steps["Run upgrade compatibility tests"]
+        # Explicit bash enables pipefail in Actions, so tee cannot hide a failure.
+        self.assertIn("        shell: bash", self.step)
+        self.assertNotIn("    strategy:", self.job)
+        rust = (ROOT / "crates/e2e_test/src/upgrade_compatibility_test.rs").read_text()
+        self.expected = set(re.findall(r'#\[ignore[^\n]*\]\s*async fn (\w+)\(', rust))
+        self.assertTrue(self.expected)
+
+    def run_cases(self, failed_test="", fail_tee=False):
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            commands = root / "commands.jsonl"
+            fake_python = root / "python3"
+            fake_python.write_text(f"#!{sys.executable}\n" + '''import json, os, sys
+from pathlib import Path
+args = sys.argv[1:]
+assert args[:8] == ["scripts/e2e_binary.py", "run", "--", "cargo", "test", "--locked", "-p", "e2e_test"]
+assert args[9:] == ["--", "--ignored", "--exact", "--nocapture"]
+test = args[8].removeprefix("upgrade_compatibility_test::")
+assert args[8] != test
+assert os.environ["no_proxy"] == "127.0.0.1,localhost,::1"
+with open(os.environ["COMMANDS"], "a") as output:
+    output.write(json.dumps({"test": test, "logs": os.environ["RUSTFS_E2E_LOG_DIR"],
+                             "evidence": os.environ["RUSTFS_SCANNER_HEAL_G09_EVIDENCE_DIR"]}) + "\\n")
+Path(os.environ["RUSTFS_SCANNER_HEAL_G09_EVIDENCE_DIR"]).mkdir(parents=True)
+print("executed " + test)
+sys.exit(17 if test == os.environ["FAILED_TEST"] else 0)
+''')
+            fake_python.chmod(0o755)
+            if fail_tee:
+                tee = root / "tee"
+                tee.write_text("#!/bin/sh\ncat >/dev/null\nexit 23\n")
+                tee.chmod(0o755)
+            env = dict(os.environ, PATH=f"{root}{os.pathsep}{os.environ['PATH']}",
+                       RUNNER_TEMP=str(root), COMMANDS=str(commands), FAILED_TEST=failed_test,
+                       NO_PROXY="127.0.0.1,localhost,::1")
+            result = subprocess.run(["bash", "--noprofile", "--norc", "-e", "-o", "pipefail", "-c", shell_body(self.step)],
+                                    cwd=root, env=env, capture_output=True, text=True)
+            calls = [json.loads(line) for line in commands.read_text().splitlines()]
+            self.assertEqual({call["test"] for call in calls}, self.expected)
+            self.assertEqual(len(calls), len(self.expected))
+            for field in ("logs", "evidence"):
+                self.assertEqual(len({call[field] for call in calls}), len(calls))
+                self.assertTrue(all(Path(call[field]).is_dir() for call in calls))
+            if not fail_tee:
+                for call in calls:
+                    self.assertIn(call["test"], (Path(call["logs"]) / "test.log").read_text())
+            return result
+
+    def test_all_ignored_upgrade_cases_run_once_with_separate_evidence(self):
+        result = self.run_cases()
+        self.assertEqual(result.returncode, 0, result.stderr)
+
+    def test_early_or_late_failure_does_not_hide_remaining_cases(self):
+        for name in ("direct_upgrade_from_rc2_preserves_object_contracts", "rc5_baseline_replicates_multipart_layouts"):
+            with self.subTest(name=name):
+                result = self.run_cases(failed_test=name)
+                self.assertNotEqual(result.returncode, 0, result.stdout)
+                self.assertIn(f"::error::Upgrade compatibility failed: {name}", result.stdout)
+
+    def test_log_write_failure_fails_the_job(self):
+        self.assertNotEqual(self.run_cases(fail_tee=True).returncode, 0)
+
+    def test_reports_are_uploaded_even_after_failure(self):
+        for name, directory in (("Upload scanner/heal G09 evidence", "rustfs-upgrade-g09-evidence"),
+                                ("Upload server logs", "rustfs-upgrade-logs")):
+            step = self.steps[name]
+            self.assertIn("        if: always()", step)
+            self.assertIn("          path: ${{ runner.temp }}/" + directory, step)
+
+
+if __name__ == "__main__":
+    unittest.main()