diff --git a/scripts/ci/warm_distance.py b/scripts/ci/warm_distance.py index 1bcee752db9a..6fc2ec567a63 100644 --- a/scripts/ci/warm_distance.py +++ b/scripts/ci/warm_distance.py @@ -216,6 +216,7 @@ def start_distance(current: Mapping[str, list], start: Mapping[str, list] | None _deadline: list[float] = [float("inf")] GIT_TIMEOUT_SECONDS = 10 FETCH_TIMEOUT_SECONDS = 20 +FETCH_RESERVE_SECONDS = 2 # of the picker's budget, kept for the diffs after the base fetch RECORD_BUDGET_SECONDS = 60 PICKER_BUDGET_SECONDS = 8 @@ -624,7 +625,7 @@ def holding(key: str) -> set[str]: HOOK_MAX_FILES = 400 # Distinct kept merge bases compared per decision (one tree diff each; one fetch brings all that are missing). MAX_ROUTE_BASES = 24 -DISTANCE_BUDGET_SECONDS = 15 +DISTANCE_BUDGET_SECONDS = 30 # two fetch attempts of ~14 s: a checkout fetch took 11.5 s on a slow runner WARM_SHA = re.compile(r"[0-9a-f]{40}") @@ -718,12 +719,42 @@ def hook_root_cost(changes: tuple[set[str], bool] | None, stamp: Mapping[str, An return (model.get("kept") or {}).get(name, model["tiers"][name]), name, count -def fetch_bases(workspace: Path, shas: Iterable[str]) -> None: - """The commits and trees (no blobs) of SHAS the checkout lacks, in one shallow fetch.""" +def fetch_bases(workspace: Path, shas: Iterable[str]) -> dict[str, Any]: + """The commits and trees (no blobs) of SHAS the checkout lacks, in one shallow fetch, tried twice. + + On 2026-09-27 the one fetch failed on some picker runs (1 of 14 bases compared), and every root of + every mini then cost the unknown start, so distance routing never pinned. A second attempt takes + what the first left missing. Returns what happened, for the decision record: the bases missing, + the seconds, and the last failure's stderr tail.""" missing = sorted({sha for sha in shas if WARM_SHA.fullmatch(sha) and not have_commit(workspace, sha)}) - if missing: - git(workspace, "fetch", "--quiet", "--no-tags", "--no-write-fetch-head", "--depth=1", "--filter=blob:none", - "origin", *missing, timeout=FETCH_TIMEOUT_SECONDS) + report: dict[str, Any] = {"missing": len(missing), "attempts": 0} + started = time.monotonic() + env = {**os.environ, "GIT_NO_LAZY_FETCH": "1", "GIT_TERMINAL_PROMPT": "0"} + for attempt in range(2): + if not missing: + break + # Leave room for a second attempt and the diffs (tens of milliseconds each) after a slow failure. + timeout = min(FETCH_TIMEOUT_SECONDS, (_deadline[0] - time.monotonic() - FETCH_RESERVE_SECONDS) / (2 - attempt)) + if timeout <= 0: + report["error"] = "no time left" + break + report["attempts"] += 1 + try: + result = subprocess.run(["git", "-C", str(workspace), "fetch", "--quiet", "--no-tags", + "--no-write-fetch-head", "--depth=1", "--filter=blob:none", "origin", *missing], + capture_output=True, text=True, timeout=timeout, env=env) + if result.returncode != 0: + report["error"] = f"exit {result.returncode}: {result.stderr.strip()[-300:]}" + else: + report.pop("error", None) + except subprocess.TimeoutExpired: + report["error"] = f"timed out after {timeout:.0f} s" + except OSError as error: + report["error"] = f"{type(error).__name__}: {error}"[:300] + missing = [sha for sha in missing if not have_commit(workspace, sha)] + report["left"] = len(missing) + report["seconds"] = round(time.monotonic() - started, 1) + return report def main_changes(workspace: Path, old: str, new: str) -> tuple[set[str], bool] | None: @@ -875,6 +906,7 @@ def picker_distance_route(runners: Sequence[Mapping[str, Any]], root: str, *, me base = (merged_onto or "").strip().lower() number = int(pr_number) if (pr_number or "").strip().isdigit() else None diffs: dict[str, tuple[set[str], bool] | None] = {} + fetched: dict[str, Any] = {} _deadline[0] = time.monotonic() + DISTANCE_BUDGET_SECONDS try: own_files = pull_request_files(workspace, base, fetch=False) if base else None @@ -886,7 +918,7 @@ def picker_distance_route(runners: Sequence[Mapping[str, Any]], root: str, *, me bases = [*sorted(parked_bases - {"", base}), *bases][:MAX_ROUTE_BASES] if WARM_SHA.fullmatch(base): if bases: - fetch_bases(workspace, bases) + fetched = fetch_bases(workspace, bases) diffs = {onto: main_changes(workspace, onto, base) for onto in bases} diffs[base] = (set(), False) finally: @@ -912,13 +944,15 @@ def picker_distance_route(runners: Sequence[Mapping[str, Any]], root: str, *, me now=now, max_wait=routed_wait_limit(queue_rounds), runner_label=runner_label, member=member) decision["job_tier"] = job_tier - decision["bases"] = {"compared": sum(1 for value in diffs.values() if value is not None), "total": len(diffs)} + decision["bases"] = {"compared": sum(1 for value in diffs.values() if value is not None), "total": len(diffs), + "fetch": fetched} return (json.dumps([root, runner_label(name)], separators=(",", ":")) if name else ""), decision def route_record(decision: Mapping[str, Any]) -> dict[str, Any]: """The picker's decision as admission records it (`route.picker`, for ci-dash's Estimates view), bounded: - mode, chosen runner ("" for the root label), predicted and baseline compile seconds, and the candidates.""" + mode, chosen runner ("" for the root label), predicted and baseline compile seconds, the candidates, and + (distance mode) how many kept merge bases it could compare and how their fetch went.""" candidates = [] for row in (decision.get("candidates") or [])[:12]: if isinstance(row, Mapping): @@ -934,7 +968,16 @@ def route_record(decision: Mapping[str, Any]) -> dict[str, Any]: predicted = picked[0]["cost"] if picked else decision.get("baseline_seconds") return {"mode": decision.get("mode") or "key", "chosen": chosen, "predicted": predicted, "baseline": decision.get("baseline", decision.get("baseline_seconds")), "tier": decision.get("tier"), - "job_tier": decision.get("job_tier"), "candidates": candidates, "why": str(decision.get("why") or "")[:200]} + "job_tier": decision.get("job_tier"), "candidates": candidates, "why": str(decision.get("why") or "")[:200], + **({"bases": bases_record(decision["bases"])} if isinstance(decision.get("bases"), Mapping) else {})} + + +def bases_record(bases: Mapping[str, Any]) -> dict[str, Any]: + """The decision's base comparison, bounded: compared of total, and the fetch's outcome.""" + fetch = bases.get("fetch") if isinstance(bases.get("fetch"), Mapping) else {} + return {"compared": bases.get("compared"), "total": bases.get("total"), + "fetch": {key: (str(fetch[key])[:160] if key == "error" else fetch[key]) + for key in ("missing", "left", "attempts", "seconds", "error") if key in fetch}} # Fitting ---------------------------------------------------------------------------------------------------- diff --git a/tests/test_ci_pr_runner_pool.py b/tests/test_ci_pr_runner_pool.py index 773f1c76e658..bb6b5a55d4ca 100644 --- a/tests/test_ci_pr_runner_pool.py +++ b/tests/test_ci_pr_runner_pool.py @@ -2002,7 +2002,7 @@ def outputs(self, runners, *, merged_onto=MERGE_BASE, slots='{"std": 40, "root-s unittest.mock.patch.object(pool.GitHub, "runners", return_value=runners), \ unittest.mock.patch.object(warm_distance, "load_model", return_value=ROUTE_MODEL), \ unittest.mock.patch.object(pool.GitHub, "get", return_value={"artifacts": []}), \ - unittest.mock.patch.object(warm_distance, "fetch_bases"), \ + unittest.mock.patch.object(warm_distance, "fetch_bases", return_value={}), \ unittest.mock.patch.object(warm_distance, "main_changes", side_effect=lambda _, old, new: (changes or {}).get(old)), \ unittest.mock.patch("sys.stdout", io.StringIO()): diff --git a/tests/test_ci_warm_distance.py b/tests/test_ci_warm_distance.py index bff46dd96f05..b5c9591703f1 100644 --- a/tests/test_ci_warm_distance.py +++ b/tests/test_ci_warm_distance.py @@ -10,6 +10,7 @@ import subprocess import sys import tempfile +import time import unittest import unittest.mock from pathlib import Path @@ -479,6 +480,29 @@ def test_this_pull_requests_parked_build_draws_its_package_change_to_its_mini(se self.assertEqual(wd.own_parked(minis["m2"][0], 7), [parked]) self.assertEqual(wd.own_parked(minis["m2"][0], None), []) + def test_the_base_fetch_is_tried_twice_and_reported(self): + with tempfile.TemporaryDirectory() as tmp: + workspace = Path(tmp) + subprocess.run(["git", "init", "-q", tmp], check=True) + calls = [] + + def fake_run(args, **_kwargs): + calls.append(args) + return subprocess.CompletedProcess(args, 128, stdout="", stderr="fatal: the remote hung up") + with unittest.mock.patch.object(wd.subprocess, "run", side_effect=fake_run), \ + unittest.mock.patch.object(wd, "have_commit", return_value=False): + wd._deadline[0] = time.monotonic() + 30 + try: + report = wd.fetch_bases(workspace, ["1" * 40, "2" * 40, "not-a-sha"]) + finally: + wd._deadline[0] = float("inf") + self.assertEqual(len(calls), 2) + self.assertEqual({key: report[key] for key in ("missing", "attempts", "left")}, + {"missing": 2, "attempts": 2, "left": 2}) + self.assertIn("remote hung up", report["error"]) + record = wd.bases_record({"compared": 1, "total": 3, "fetch": {**report, "error": "x" * 999}}) + self.assertEqual((record["compared"], len(record["fetch"]["error"])), (1, 160)) + def test_record_is_bounded_and_reads_either_mode(self): minis = {"m1": [self.stamp("1" * 40)]} _, decision = self.route([runner("m1-glaeda")], minis, {"1" * 40: (swift(1), False)})