feat(ci): measure queue and execution time by run attempt (#7689)

* feat(ci): measure queue and execution time by run attempt

* fix(ci): bound timing samples and recognize the old workspace lane

* fix(ci): sample completed runs for stable timing comparisons

.config/make/tests.mak

@@ -55,6 +55,7 @@ script-tests: ## Run shell script tests
 	$(RUSTFS_PYTHON_BIN) ./scripts/check_scheduled_validation_freshness.py --self-test
 	$(RUSTFS_PYTHON_BIN) ./scripts/test_security_workflow.py
 	$(RUSTFS_PYTHON_BIN) ./scripts/test_nightly_candidate.py
+	$(RUSTFS_PYTHON_BIN) ./scripts/test_ci_timing_report.py
 	$(RUSTFS_PYTHON_BIN) ./scripts/s3-tests/test_report_compat.py
 	bash -n ./scripts/validate_object_data_cache_cold_stampede.sh
 	$(RUSTFS_PYTHON_BIN) ./scripts/check_object_data_cache_follower_samples.py --self-test

docs/testing/ci-timing.md

@@ -0,0 +1,9 @@
+# CI timing samples
+
+Collect a bounded sample of completed PR CI runs created in the last seven days with `python3 scripts/ci_timing_report.py --limit 30 --output /tmp/ci-timing.json`. This requires an authenticated `gh` CLI with Actions read access. The JSON retains run SHA, attempt, job and step timestamps so another reviewer can reproduce the summary with `--input /tmp/ci-timing.json --output /tmp/ci-timing-summary.json` without GitHub access.
+
+The report separates job creation-to-start wait, job execution, and individual step durations. The successful-code sample excludes documentation-only runs and successful PR-closure cancellation handlers. It uses the selected attempt's start for reruns, includes parallel jobs in the runner-minute sum, and leaves missing timestamps unknown. It does not estimate compiler time from a combined build-and-test step or treat runner minutes as wall time or a bill.
+
+A small recent sample may contain no complete successful code runs. In that case its median is absent, not zero. Keep cancelled and failed counts visible; do not replace the sample with only green runs when evaluating changes.
+
+Before reducing a PR lane, compare a proposed path classifier in shadow mode against the existing full selection. Keep that lane required until equivalent nightly evidence is complete and current for the same source policy. Timing data alone does not establish functional coverage, escaped-regression rate or quarantine health. The 30–45 minute PR target in backlog #2483 remains an experiment to measure, not an acceptance result from this tool.

scripts/ci_timing_report.py

@@ -0,0 +1,148 @@
+#!/usr/bin/env python3
+"""Collect bounded Actions timing samples without changing CI selection."""
+
+from __future__ import annotations
+
+import argparse
+from collections import Counter, defaultdict
+from datetime import datetime, timedelta, timezone
+import json
+from pathlib import Path
+import statistics
+import subprocess
+from urllib.parse import urlencode
+
+
+def timestamp(value):
+    if not value:
+        return None
+    result = datetime.fromisoformat(value.replace("Z", "+00:00"))
+    if result.tzinfo is None:
+        raise ValueError("Actions timestamp has no timezone")
+    return result
+
+
+def minutes(start, end):
+    start, end = timestamp(start), timestamp(end)
+    if start is None or end is None or end < start:
+        return None
+    return (end - start).total_seconds() / 60
+
+
+def distribution(values):
+    known = [value for value in values if value is not None]
+    return {"samples": len(values), "known": len(known), "missing": len(values) - len(known),
+            "median_minutes": statistics.median(known) if known else None}
+
+
+def summarize(runs):
+    statuses = Counter()
+    jobs_by_name = defaultdict(list)
+    successful_code_runs = []
+    for run in runs:
+        statuses[run.get("conclusion") or run.get("status", "unknown")] += 1
+        jobs = run["jobs"]
+        for job in jobs:
+            if job.get("conclusion") != "skipped":
+                jobs_by_name[job["name"]].append(job)
+        # Exclude docs-only and pull_request.closed cancellation-handler greens.
+        if (run.get("status") == "completed" and run.get("conclusion") == "success" and jobs
+                and any(job["name"] in ("Workspace Test and Lint", "Test and Lint") and job.get("conclusion") == "success"
+                        and any(step.get("name") == "Run nextest tests" and step.get("conclusion") == "success" for step in (job.get("steps") or [])) for job in jobs)
+                and all(job.get("status") == "completed" for job in jobs)):
+            successful_code_runs.append(run)
+
+    wall, job_time = [], []
+    for run in successful_code_runs:
+        ends = [timestamp(job.get("completed_at")) for job in run["jobs"] if job.get("conclusion") != "skipped"]
+        end = max(ends).isoformat() if ends and all(ends) else None
+        start = run.get("run_started_at") if run.get("run_attempt", 1) > 1 else run.get("created_at")
+        wall.append(minutes(start, end))
+        durations = [minutes(job.get("started_at"), job.get("completed_at")) for job in run["jobs"] if job.get("conclusion") != "skipped"]
+        job_time.append(sum(durations) if durations and all(value is not None for value in durations) else None)
+
+    job_summary = {}
+    for name, jobs in sorted(jobs_by_name.items()):
+        steps = defaultdict(list)
+        for job in jobs:
+            for step in (job.get("steps") or []):
+                if step.get("conclusion") != "skipped":
+                    duration = minutes(step.get("started_at"), step.get("completed_at")) if step.get("status") == "completed" else None
+                    steps[step["name"]].append(duration)
+        job_summary[name] = {
+            "queue": distribution([minutes(job.get("created_at"), job.get("started_at")) for job in jobs]),
+            "execution": distribution([minutes(job.get("started_at"), job.get("completed_at")) if job.get("status") == "completed" else None for job in jobs]),
+            "steps": {name: distribution(values) for name, values in sorted(steps.items())},
+        }
+    total = len(runs)
+    return {"runs": total, "statuses": dict(statuses),
+            "cancelled_fraction": statuses["cancelled"] / total if total else None,
+            "completed_fraction": sum(run.get("status") == "completed" for run in runs) / total if total else None,
+            "successful_code_runs": len(successful_code_runs),
+            "successful_code_wall": distribution(wall), "successful_code_job_sum": distribution(job_time),
+            "jobs": job_summary,
+            "limits": ["Live collection samples completed runs; ongoing queue depth is not measured.",
+                       "Job creation-to-start is scheduler wait; dependency delay is not included.",
+                       "Step times can combine setup, compilation and tests; they do not isolate compiler time.",
+                       "Job sums are unweighted runner minutes, not billing or wall time.",
+                       "Actions timing does not establish functional completeness, escaped regressions or quarantine health."]}
+
+
+def api(path):
+    result = subprocess.run(["gh", "api", path], check=True, capture_output=True, text=True, timeout=60)
+    return json.loads(result.stdout)
+
+
+def collect(repository, limit, since=None):
+    since = since or datetime.now(timezone.utc) - timedelta(days=7)
+    runs = []
+    page = 1
+    while len(runs) < limit:
+        query = urlencode({"event": "pull_request", "status": "completed", "per_page": 100, "page": page, "created": ">=" + since.isoformat()})
+        batch = api(f"repos/{repository}/actions/workflows/ci.yml/runs?{query}")["workflow_runs"]
+        if any(timestamp(run.get("created_at")) is None or timestamp(run["created_at"]) < since for run in batch):
+            raise ValueError("GitHub returned runs outside the requested date range")
+        runs.extend(batch[:limit - len(runs)])
+        if len(batch) < 100:
+            break
+        page += 1
+    samples = []
+    for run in runs:
+        attempt = run["run_attempt"]
+        endpoint = f"repos/{repository}/actions/runs/{run['id']}/attempts/{attempt}"
+        jobs, page = [], 1
+        while True:
+            batch = api(f"{endpoint}/jobs?per_page=100&page={page}")["jobs"]
+            jobs.extend(batch)
+            if len(batch) < 100:
+                break
+            page += 1
+        # A rerun or completion while collecting must not mix snapshots.
+        observed = api(f"repos/{repository}/actions/runs/{run['id']}")
+        if any(observed.get(key) != run.get(key) for key in ("run_attempt", "head_sha", "status", "conclusion")):
+            raise ValueError(f"run {run['id']} changed during collection; collect a fresh snapshot")
+        sample = {key: run.get(key) for key in ("id", "run_attempt", "head_sha", "created_at", "run_started_at", "status", "conclusion", "html_url")}
+        sample["jobs"] = [{key: job.get(key) for key in ("id", "name", "created_at", "started_at", "completed_at", "status", "conclusion", "steps")} for job in jobs]
+        samples.append(sample)
+    return {"schema": 1, "repository": repository, "observed_at": datetime.now(timezone.utc).isoformat(), "created_since": since.isoformat(), "runs": samples}
+
+
+def main():
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--repository", default="rustfs/rustfs")
+    parser.add_argument("--limit", type=int, default=30)
+    parser.add_argument("--input", type=Path, help="summarize a previously collected JSON sample")
+    parser.add_argument("--output", type=Path, required=True)
+    args = parser.parse_args()
+    if not 1 <= args.limit <= 200:
+        parser.error("--limit must be between 1 and 200")
+    snapshot = json.loads(args.input.read_text()) if args.input else collect(args.repository, args.limit)
+    if snapshot.get("schema") != 1 or not isinstance(snapshot.get("runs"), list):
+        parser.error("unsupported timing snapshot")
+    snapshot["summary"] = summarize(snapshot["runs"])
+    args.output.write_text(json.dumps(snapshot, indent=2) + "\n")
+    print(json.dumps({key: value for key, value in snapshot["summary"].items() if key != "jobs"}, indent=2))
+
+
+if __name__ == "__main__":
+    main()

scripts/test_ci_timing_report.py

@@ -0,0 +1,74 @@
+#!/usr/bin/env python3
+"""Timing reports must preserve missing evidence and non-successful runs."""
+import copy
+from datetime import datetime, timezone
+import unittest
+from unittest import mock
+
+from ci_timing_report import collect, summarize
+
+
+class TimingTests(unittest.TestCase):
+    def setUp(self):
+        self.job = {"id": 2, "name": "Workspace Test and Lint", "status": "completed", "conclusion": "success",
+                    "created_at": "2026-09-01T00:10:00Z", "started_at": "2026-09-01T00:15:00Z", "completed_at": "2026-09-01T00:25:00Z",
+                    "steps": [{"name": "Run nextest tests", "status": "completed", "conclusion": "success", "started_at": "2026-09-01T00:16:00Z", "completed_at": "2026-09-01T00:24:00Z"}]}
+        self.run = {"id": 1, "run_attempt": 2, "head_sha": "a" * 40, "status": "completed", "conclusion": "success",
+                    "created_at": "2026-09-01T00:00:00Z", "run_started_at": "2026-09-01T00:00:00Z", "jobs": [self.job]}
+
+    def test_parallel_minutes_are_not_wall_time(self):
+        self.run["jobs"].append({**self.job, "id": 3, "name": "Other"})
+        result = summarize([self.run])
+        self.assertEqual(result["successful_code_wall"]["median_minutes"], 25)
+        self.assertEqual(result["successful_code_job_sum"]["median_minutes"], 20)
+        self.assertEqual(result["jobs"][self.job["name"]]["queue"]["median_minutes"], 5)
+        self.assertEqual(result["jobs"][self.job["name"]]["steps"]["Run nextest tests"]["median_minutes"], 8)
+
+    def test_cancelled_incomplete_and_docs_runs_are_not_success_samples(self):
+        cancelled = {**self.run, "status": "completed", "conclusion": "cancelled"}
+        pending = {**self.run, "status": "queued", "conclusion": None, "jobs": []}
+        docs = {**self.run, "jobs": [{**self.job, "name": "Quick Checks"}]}
+        result = summarize([self.run, cancelled, pending, docs])
+        self.assertEqual(result["successful_code_runs"], 1)
+        self.assertEqual(result["cancelled_fraction"], 0.25)
+        self.assertEqual(result["completed_fraction"], 0.75)
+
+    def test_rerun_wall_time_excludes_time_before_the_attempt(self):
+        self.run["created_at"] = "2026-08-01T00:00:00Z"
+        self.assertEqual(summarize([self.run])["successful_code_wall"]["median_minutes"], 25)
+        del self.run["run_started_at"]
+        self.assertIsNone(summarize([self.run])["successful_code_wall"]["median_minutes"])
+
+    def test_old_workspace_job_name_requires_an_executed_test_step(self):
+        self.job["name"] = "Test and Lint"
+        self.assertEqual(summarize([self.run])["successful_code_runs"], 1)
+        self.job["steps"] = []
+        self.assertEqual(summarize([self.run])["successful_code_runs"], 0)
+
+    def test_collection_rejects_out_of_range_api_results(self):
+        with mock.patch("ci_timing_report.api", return_value={"workflow_runs": [self.run]}):
+            with self.assertRaisesRegex(ValueError, "outside the requested date range"):
+                collect("owner/repo", 1, datetime(2026, 9, 2, tzinfo=timezone.utc))
+
+    def test_missing_times_are_unknown_not_zero(self):
+        self.job["created_at"] = None
+        self.job["completed_at"] = None
+        result = summarize([self.run])
+        self.assertIsNone(result["successful_code_wall"]["median_minutes"])
+        self.assertIsNone(result["successful_code_job_sum"]["median_minutes"])
+        self.assertEqual(result["jobs"][self.job["name"]]["queue"]["missing"], 1)
+
+    def test_collection_binds_the_attempt_and_rejects_reruns(self):
+        source = copy.deepcopy(self.run)
+        del source["jobs"]
+        with mock.patch("ci_timing_report.api", side_effect=[{"workflow_runs": [source]}, {"jobs": [self.job]}, source]) as api:
+            snapshot = collect("owner/repo", 1, datetime(2026, 8, 1, tzinfo=timezone.utc))
+            self.assertIn("/attempts/2/jobs?", api.call_args_list[1].args[0])
+            self.assertEqual(snapshot["runs"][0]["run_attempt"], 2)
+        with mock.patch("ci_timing_report.api", side_effect=[{"workflow_runs": [source]}, {"jobs": [self.job]}, {**source, "run_attempt": 3}]):
+            with self.assertRaisesRegex(ValueError, "changed during collection"):
+                collect("owner/repo", 1, datetime(2026, 8, 1, tzinfo=timezone.utc))
+
+
+if __name__ == "__main__":
+    unittest.main()