diff --git a/.github/workflows/test-e2e.yml b/.github/workflows/test-e2e.yml index 1886d41ce4fd..309f050e22f0 100644 --- a/.github/workflows/test-e2e.yml +++ b/.github/workflows/test-e2e.yml @@ -74,7 +74,7 @@ jobs: # CI_E2E_LARGE_POOL_OVERFLOW=0 turns the choice off. run-e2e.sh applies the # same rule from the same script. With CI_PR_POOL_OWNED=1 an owned Mac # (glaeda-*) comes first while its job starts within CI_PR_POOL_QUEUE_ROUNDS - # job lengths and no later than on Blacksmith, as for pull requests. A job an owned Mac leaves queued or refuses is re-run by + # job lengths, whatever Blacksmith's wait, as for pull requests. A job an owned Mac leaves queued or refuses is re-run by # ci-owned-pool-rescue.yml, and every re-run attempt of build or test takes # retry_label (Blacksmith) instead, so a busy Mac never strands a run. runs-on: ${{ github.repository_owner != 'manaflow-ai' && 'ubuntu-24.04' || vars.LINUX_RUNNER || 'blacksmith-4vcpu-ubuntu-2404' }} diff --git a/docs/ci-runners.md b/docs/ci-runners.md index 5f1383399ea8..f381d2d1fb22 100644 --- a/docs/ci-runners.md +++ b/docs/ci-runners.md @@ -134,10 +134,14 @@ linked it. The run takes the pool in `CI_PR_POOL_ORDER` with the least expected wait and no queued release or nightly job: the queue its job joins in rounds (queued jobs over capacity) times a job's length there (5 minutes on 12vcpu, 10 elsewhere). Owned pools come first and take the run while its -jobs start there no later than on the best Blacksmith pool, within -`CI_PR_POOL_QUEUE_ROUNDS` job lengths, and while everything the runs holding -the pool will need at their peak, plus this run's, stays within machines x -(1 + rounds). A pool's +jobs start there within `CI_PR_POOL_QUEUE_ROUNDS` job lengths, whatever +Blacksmith's wait (Blacksmith is overflow), and while the queue stays within +machines x (1 + rounds). With the runners read live, that counts the busy +runners, the janitor's queue and the peaks of the runs of the last 10 +minutes; without them, everything the runs holding the pool will need at +their peak. With live runners, a missing or stale snapshot no longer skips +the fleet: the owned pools are decided live, and a run none takes keeps its +default route. A pool's capacity is what it ran at most while jobs queued behind it (`POOL_CAPACITIES`): 5 for 12vcpu, 10 for each 6vcpu pool. At 23:16Z on 2026-09-24, counted at 10, 12vcpu ran 3 with 18 queued while macOS 15 ran 1 @@ -151,7 +155,7 @@ run there compiles cold, 10 to 20 minutes longer, about one job's length. | `CI_PR_POOL_OVERFLOW` | unset (on) | `0` turns the preference off; every job takes its `MACOS_RUNNER_PR` route | | `CI_PR_POOL_ORDER` | `blacksmith-12vcpu-macos-26,blacksmith-6vcpu-macos-26,blacksmith-6vcpu-macos-15` | preference order; only pools whose Xcode pin `pr_runner_pool.py` knows are accepted, and an unknown label turns the preference off | | `CI_PR_POOL_MAX_QUEUED` | `0` | with `CI_PR_POOL_QUEUE_ROUNDS=0` only: a Blacksmith pool still takes a run with up to this many macOS jobs queued once it arrives | -| `CI_PR_POOL_QUEUE_ROUNDS` | `1` | the most job lengths a run's jobs may expect to wait on an owned pool (at most `3`); within that they queue there while they would start no later than on Blacksmith, and the peaks of the runs holding it stay within machines x (1 + rounds). `0` is the kill switch and restores the old rule exactly: an owned pool only when the run's peak is free counting every run's peak, and a full Blacksmith pool rolls over at once | +| `CI_PR_POOL_QUEUE_ROUNDS` | `1` | the most job lengths a run's jobs may expect to wait on an owned pool (at most `3`); within that they queue there whatever Blacksmith's wait, while the queue stays within machines x (1 + rounds). `0` is the kill switch and restores the old rule exactly: an owned pool only when the run's peak is free counting every run's peak, and a full Blacksmith pool rolls over at once | The two macOS 26 pools share the lane's Xcode. A run on `blacksmith-6vcpu-macos-15` builds with `CMUX_CI_XCODE_APP_MACOS_15`, the pool diff --git a/scripts/ci/e2e_runner_pool.py b/scripts/ci/e2e_runner_pool.py index bf2f1f687e2e..0dba884aa17d 100644 --- a/scripts/ci/e2e_runner_pool.py +++ b/scripts/ci/e2e_runner_pool.py @@ -52,8 +52,8 @@ (vars.CMUX_CI_XCODE_APP_PR), ahead of Blacksmith, on a snapshot younger than pr_runner_pool.MAX_SNAPSHOT_MINUTES, and by pull request CI's queue rule (vars.CI_PR_POOL_QUEUE_ROUNDS, `--queue-rounds`): an owned pool takes the run -while its job starts there within that many job lengths and no later than on -the best Blacksmith pool, and the queue stays within machines x (1 + rounds) +while its job starts there within that many job lengths, whatever +Blacksmith's wait, and the queue stays within machines x (1 + rounds) (pr_runner_pool.owned_room()). Without the rounds the picker used the kill switch rule, which counts every in-flight run's whole future peak (the janitor's `committed`) as taken now: on 2026-09-25 that read 43 of 32 std diff --git a/scripts/ci/ios_runner_pool.py b/scripts/ci/ios_runner_pool.py index 1a8b64ef811e..17c931077cb2 100644 --- a/scripts/ci/ios_runner_pool.py +++ b/scripts/ci/ios_runner_pool.py @@ -40,8 +40,8 @@ Queue rounds. The pool rule is pull request CI's with its vars.CI_PR_POOL_QUEUE_ROUNDS (`--queue-rounds`): an owned pool takes the run -while its jobs start there within that many job lengths and no later than on -the best Blacksmith pool, and the queue stays within machines x (1 + rounds) +while its jobs start there within that many job lengths, whatever +Blacksmith's wait, and the queue stays within machines x (1 + rounds) (pr_runner_pool.owned_room()). Without the rounds the picker used the kill switch rule, which counts every in-flight run's whole future peak (the janitor's `committed`) as taken now: on 2026-09-25 (run 36136190497) that read diff --git a/scripts/ci/pr_runner_pool.py b/scripts/ci/pr_runner_pool.py index 31eec36deaf8..fdad24436752 100644 --- a/scripts/ci/pr_runner_pool.py +++ b/scripts/ci/pr_runner_pool.py @@ -24,11 +24,13 @@ DerivedData seed for its Xcode and compiles cold Owned pools come first in the default order and take the run while the jobs -they would hold start no later than this run's jobs would on the best -Blacksmith pool, within vars.CI_PR_POOL_QUEUE_ROUNDS job lengths (default 1, -at most MAX_QUEUE_ROUNDS), and within the queue bound (owned_room()). -Otherwise the Blacksmith pool with the least expected wait takes it, the -earlier in the order on a tie. +they would hold start within vars.CI_PR_POOL_QUEUE_ROUNDS job lengths +(default 1, at most MAX_QUEUE_ROUNDS) and within the queue bound +(owned_room()), whatever Blacksmith's expected wait: Blacksmith is overflow. +Comparing with that wait, read from a snapshot up to MAX_SNAPSHOT_MINUTES old, +sent runs off minis busy for a few minutes whenever a Blacksmith pool had +looked free. Otherwise the Blacksmith pool with the least expected wait takes +it, the earlier in the order on a tie. The wait counts what holds a label now: jobs queued and running, and each run since the snapshot at what it holds (young_charge(): admission and its @@ -83,7 +85,15 @@ gives the online runners carrying each label, its capacity, and the idle ones among them; every other online runner counts as busy (live_pools()); a label with no idle runner is charged the -snapshot's queue and the runs since it, since the API shows no queue. +snapshot's queue and one job per run since it, since the API shows no queue. +Read live, in-flight runs' peaks (`committed`, the markers' peaks beyond the +live window) are not charged at all: those are the fallback without the +runners API. With the runners read, attempt 1 does not need the snapshot +either: when it cannot be downloaded or is stale, the owned pools are +decided from the runners alone and Blacksmith's queues count as unknown +(empty): a run that no owned pool takes keeps every job's default, as +before. Read live, the queue bound counts the peaks of the runs of the last +DEFAULT_JOB_MINUTES that took the pool (their shards are on the way). A job on an owned pool may therefore wait up to about CI_PR_POOL_QUEUE_ROUNDS job lengths, and ci-owned-pool-rescue.yml gives a CI run's jobs that much (QUEUE_ROUND_SECONDS per round) on top of its budget before it moves the @@ -508,11 +518,25 @@ class Routed: pick already finished without a marker, so they hold no owned machine. `unknown` counts runs whose pick this one cannot see yet; they are replayed and charged REPLAYED_RUN_JOBS on an owned pool they could take. + `owned_runs` counts the runs behind `owned` on each pool. `live_now` and + `live_runs` are `owned_now` and `owned_runs` for the runs younger than + LIVE_WINDOW_MINUTES alone (None: not split by age). """ unknown: int = 0 owned: Mapping[str, int] = dataclasses.field(default_factory=dict) ephemeral: int = 0 owned_now: Mapping[str, int] | None = dataclasses.field(default=None, compare=False) # None: `owned` + # None: one run per pool with a peak in `owned`. + owned_runs: Mapping[str, int] | None = dataclasses.field(default=None, compare=False) + live_now: Mapping[str, int] | None = dataclasses.field(default=None, compare=False) + live_runs: Mapping[str, int] | None = dataclasses.field(default=None, compare=False) + # `unknown` for the runs younger than LIVE_WINDOW_MINUTES alone (None: not split). + live_unknown: int | None = dataclasses.field(default=None, compare=False) + + def runs(self) -> Mapping[str, int]: + if self.owned_runs is not None: + return self.owned_runs + return {label: 1 for label, peak in self.owned.items() if peak} def flag(value: str | None) -> bool: @@ -955,6 +979,18 @@ def owned_room(label: str, counts: Mapping[str, int], added_jobs: int, taken_pea return min(wait, bound) +def snapshot_problem(snapshot: Any, now: dt.datetime) -> str: + """Why decide() would not read `snapshot` (missing, undated, stale or without pools), or "".""" + if not isinstance(snapshot, Mapping) or not snapshot.get("generated_at"): + return "no readable pool snapshot" + if not isinstance(snapshot.get("pools"), Mapping): + return "malformed pool snapshot" + age = snapshot_age_minutes(snapshot, now) + if age is None or age < -5 or age > MAX_SNAPSHOT_MINUTES: + return f"pool snapshot is stale or undated (age {age if age is None else round(age)} min)" + return "" + + def iso(moment: dt.datetime) -> str: return moment.astimezone(dt.timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") @@ -1136,11 +1172,17 @@ def live_pools(snapshot: Mapping[str, Any], idle: Mapping[str, int], slot_counts an offline machine counts as running. What is not idle is running. The runners API shows no queue: a label with an idle runner has none, and one without counts the - snapshot's queue plus the peaks of the runs that took the pool since the - snapshot and before the live window (`older`), which errs high. - `future`, for the queue bound only (owned_room()), is what the runs - holding the label will need at their peak: the snapshot's `committed` - plus those older peaks, since the API cannot see jobs not created yet. + snapshot's queue plus `older`: the jobs of the runs that took the pool + since the snapshot and before the live window that may still wait there + (pull request CI passes one per run, its admission). + + The snapshot's `committed` (every in-flight run's peak) is not charged: + charging it kept runs off idle minis ("0 of 19 root runners free" and + "every pool is full" while 12.7k unit-min of owned time sat idle beside + Blacksmith work on 2026-09-25). It remains the fallback when the runners + cannot be read. `future` is 0, so the queue bound (owned_room()) counts + what holds the label now plus what choose() passes: the peaks of runs + younger than DEFAULT_JOB_MINUTES, less what they already hold. A root label counts only while CI_OWNED_POOL_SLOTS gives it a root count, which is what turns root routing on (root_label()). @@ -1156,10 +1198,8 @@ def live_pools(snapshot: Mapping[str, Any], idle: Mapping[str, int], slot_counts else: capacity[label] = max(int(slot_counts.get(label) or 0), free) seen = (pools.get(label) or {}) if isinstance(pools.get(label), Mapping) else {} - older_peaks = older.get(pool_label(label), 0) - queued = 0 if free else int(seen.get("queued") or 0) + older_peaks - pools[label] = {"queued": queued, "running": capacity[label] - free, "committed": 0, - "future": int(seen.get("committed") or 0) + older_peaks} + queued = 0 if free else int(seen.get("queued") or 0) + max(0, int(older.get(pool_label(label), 0))) + pools[label] = {"queued": queued, "running": capacity[label] - free, "committed": 0, "future": 0} return {**snapshot, "pools": pools}, capacity @@ -1184,21 +1224,20 @@ def pick(load: Mapping[str, Mapping[str, int]], added: Mapping[str, int], usable compared_jobs: int | None = None) -> Pick: """The rule itself. `added` counts runs replayed since the snapshot on each pool. - With `queue_rounds` (CI_PR_POOL_QUEUE_ROUNDS) above 0, the run goes where - it expects to wait least (expected_wait()): an owned pool, in order, - when the jobs it would place there start no later than this run's jobs - would on the best Blacksmith pool, within `queue_rounds` job lengths, - and within the queue bound (owned_room()); else the Blacksmith pool with - the least expected wait, the earlier in order on a tie. `taken` is the + With `queue_rounds` (CI_PR_POOL_QUEUE_ROUNDS) above 0: an owned pool, in + order, when the jobs it would place there start within `queue_rounds` + job lengths and within the queue bound (owned_room()), whatever + Blacksmith's expected wait (Blacksmith is overflow); else the Blacksmith + pool with the least expected wait (expected_wait()), the earlier in + order on a tie. `taken` is the peak of the runs since the snapshot that took each owned pool, by their markers, and `taken_now` what they hold now (young_charge()). A replayed run counts REPLAYED_RUN_JOBS on an owned pool and one on its root runners and on Blacksmith. `reserve` (main's full suite with CI_OWNED_MAIN_RESERVE) is kept free on top of this run's jobs and root jobs, and turns `split` off. `compared_jobs` (default `jobs`) is how - many jobs the owned and Blacksmith waits are compared at: a replayed run - takes the pool with one job but is compared at REPLAYED_RUN_JOBS, the - jobs it has at once. + many jobs Blacksmith's wait (reported, not compared) is read at: a + replayed run takes the pool with one job but has REPLAYED_RUN_JOBS at once. With 0, the old rule: an owned pool only with machines free now, the first Blacksmith pool with a free machine (or at most max_queued @@ -1224,7 +1263,12 @@ def pick(load: Mapping[str, Mapping[str, int]], added: Mapping[str, int], usable for label in usable: if not persistent(label): continue - limit = min(whole, queue_rounds * job_minutes(label)) if queue_rounds else 0.0 + # Fleet first: an owned pool may queue its full allowance whatever + # Blacksmith's wait looks like. That wait comes from a snapshot up to + # MAX_SNAPSHOT_MINUTES old and is 0 whenever a Blacksmith pool showed + # a free machine, which sent runs off minis busy for a few minutes + # (2026-09-25: 2.5k Blacksmith job-min on "0 of 19 root runners free"). + limit = float(queue_rounds * job_minutes(label)) peak, now = taken.get(label, 0), taken_now.get(label, 0) room = owned_room(label, load[label], added[label] * REPLAYED_RUN_JOBS, peak, now, queue_rounds, limit) root_room = (owned_room(label, roots[label], added[label], peak, now, queue_rounds, limit) @@ -1523,11 +1567,14 @@ def choose( if limits is None: return Choice("", "", f"{OVERFLOW_VARIABLE} is 0, or {ORDER_VARIABLE}/{MAX_QUEUED_VARIABLE}/" f"{QUEUE_ROUNDS_VARIABLE} is invalid"), None + unreadable = "" try: snapshot = fetch() - except Exception as error: # noqa: BLE001 - every failure keeps the default - return Choice("", "", f"could not read the pool snapshot ({error})"), None + except Exception as error: # noqa: BLE001 - every failure keeps the default, or the live runners decide + snapshot, unreadable = None, f"could not read the pool snapshot ({error})" if fork: + if unreadable: + return Choice("", "", unreadable), None # No repository variables reach a fork run; the janitor copied them. copied = snapshot.get("settings") if isinstance(snapshot, Mapping) else None if not isinstance(copied, Mapping): @@ -1559,33 +1606,68 @@ def choose( if not persistent(label) or light and label.startswith(f"glaeda-{LIGHT_CLASS}-"))) if not limits.order: return Choice("", "", f"retry attempt {run_attempt}; no ephemeral pool in the order"), snapshot + live = live_owned is not None and not fork + # Attempt 1 with the owned runners read live does not need the janitor: + # a failed download (an HTTP 503 from artifact storage) or a stale + # snapshot leaves the owned pools to the live runners and Blacksmith's + # queues unknown, instead of skipping the fleet (2026-09-25: 2.8k + # Blacksmith job-min on a stale snapshot, 0.6k on a 503). + live_only = "" + if live and not retry: + live_only = unreadable or snapshot_problem(snapshot, now) + if live_only: + # A stale snapshot's warm keys still name kept builds (warm affinity). + warm = snapshot.get("warm") if isinstance(snapshot, Mapping) else None + snapshot, unreadable = {"generated_at": iso(now), "pools": {}, "source": "live", "warm": warm}, "" + if unreadable: + return Choice("", "", unreadable), snapshot if not isinstance(snapshot, Mapping) or not snapshot.get("generated_at"): return Choice("", "", "no readable pool snapshot"), snapshot - try: - routed = count_routed(str(snapshot["generated_at"])) - except Exception as error: # noqa: BLE001 - every failure keeps the default - return Choice("", "", f"could not count runs since the snapshot ({error})"), snapshot + if live_only: + # The live window below counts the runs that matter. + routed = Routed() + else: + try: + routed = count_routed(str(snapshot["generated_at"])) + except Exception as error: # noqa: BLE001 - every failure keeps the default + return Choice("", "", f"could not count runs since the snapshot ({error})"), snapshot if not isinstance(routed, Routed): routed = Routed(unknown=int(routed)) owned_capacity = {} if fork else slots(owned_slots, xcode_pins.get(PR_XCODE_VARIABLE)) - live = live_owned is not None and not fork if live: # The idle runners replace the slot counts and the snapshot's owned - # counts. Only the runs of the last LIVE_WINDOW_MINUTES are charged to - # the owned pools; the rest of the snapshot window counts on Blacksmith. + # counts. Runs of the last DEFAULT_JOB_MINUTES that took an owned pool + # count their peaks toward the queue bound (owned_room()): their + # shards are on the way, and nothing else here sees them coming. Only + # those of the last LIVE_WINDOW_MINUTES hold machines toward the wait; + # older ones' jobs show on the runners. The rest of the snapshot + # window counts on Blacksmith. try: - recent = count_routed(iso(now - dt.timedelta(minutes=LIVE_WINDOW_MINUTES))) + recent = count_routed(iso(now - dt.timedelta(minutes=DEFAULT_JOB_MINUTES))) except Exception as error: # noqa: BLE001 - every failure keeps the default return Choice("", "", f"could not count recent runs ({error})"), snapshot if not isinstance(recent, Routed): recent = Routed(unknown=int(recent)) before = routed - routed = Routed(unknown=recent.unknown, owned=recent.owned, owned_now=recent.owned_now, - ephemeral=routed.ephemeral + max(0, routed.unknown - recent.unknown)) + # Replayed: only the unknown runs of the live window; older ones' + # jobs are on the runners already. + unknown = recent.unknown if recent.live_unknown is None else recent.live_unknown + held_now = recent.owned_now if recent.live_now is None else recent.live_now + # The bound: each run's peak less what the runs past the live window + # already hold, which the runners count busy. + on_runners = {label: max(0, (recent.owned_now or recent.owned).get(label, 0) - held_now.get(label, 0)) + for label in recent.owned} if recent.live_now is not None else {} + routed = Routed(unknown=unknown, + owned={label: max(0, peak - on_runners.get(label, 0)) for label, peak in recent.owned.items()}, + owned_now=held_now, + owned_runs=recent.runs() if recent.live_runs is None else recent.live_runs, + ephemeral=routed.ephemeral + max(0, routed.unknown - unknown)) + recent = routed # Runs since the snapshot but before the live window: their owned jobs # are running (so busy below) or still queued, which the runners API - # cannot show. - older = {label: max(0, count - recent.owned.get(label, 0)) for label, count in before.owned.items()} + # cannot show. One job each (admission) may still wait where no + # runner is idle; their later jobs are not charged (live_pools()). + older = {label: max(0, count - recent.runs().get(label, 0)) for label, count in before.runs().items()} snapshot, owned_capacity = live_pools(snapshot, live_owned or {}, owned_capacity, older, live_online) choice = decide(snapshot, limits, now=now, xcode_pins={} if fork else xcode_pins, routed_since=routed.unknown, owned_since=routed.owned, ephemeral_since=routed.ephemeral, @@ -1600,8 +1682,15 @@ def choose( choice = Choice("", "", f"main's full-suite dispatch: no owned pool fits its whole run with " f"{reserve} machine(s) and root runner(s) left free for pull requests " f"({choice.reason})") + if live_only and not persistent(choice.runner): + # Blacksmith's queues are unknown: overflow keeps every job's default + # route, as it did before the live runners could decide. + choice = Choice("", "", f"{live_only}; no owned pool has room ({choice.reason})") if live and persistent(choice.runner): choice = dataclasses.replace(choice, reason=f"{choice.reason}; owned machines read live from the runners API") + if live_only and choice.runner: + choice = dataclasses.replace(choice, reason=f"{choice.reason}; no janitor snapshot ({live_only}), so " + "Blacksmith's queues are unknown") if fork and choice.runner: choice = dataclasses.replace(choice, reason=f"fork head; {choice.reason}") if retry and choice.runner: @@ -1751,13 +1840,22 @@ def pull_request_routes_since(self, since: str, *, exclude_run_id: int | None, if routed_run(run) and run.get("id") != exclude_run_id and run.get("status") != "completed"] owned: dict[str, int] = {} owned_now: dict[str, int] = {} - ephemeral = unknown = looked_up = 0 + owned_runs: dict[str, int] = {} + live_now: dict[str, int] = {} + live_runs: dict[str, int] = {} + ephemeral = unknown = live_unknown = looked_up = 0 + clock = now or dt.datetime.now(dt.timezone.utc) + + def young(run: Mapping[str, Any]) -> bool: + age = run_age_minutes(run, clock) + return age is not None and age < LIVE_WINDOW_MINUTES for run in runs: if not may_hold_owned_pool(run, light_retry=light_retry): ephemeral += 1 continue if looked_up >= ROUTE_LOOKUPS: unknown += 1 + live_unknown += young(run) continue looked_up += 1 try: @@ -1766,13 +1864,19 @@ def pull_request_routes_since(self, since: str, *, exclude_run_id: int | None, route = None if isinstance(route, tuple): owned[route[0]] = owned.get(route[0], 0) + route[1] - owned_now[route[0]] = owned_now.get(route[0], 0) + young_charge( - route[1], run_age_minutes(run, now or dt.datetime.now(dt.timezone.utc))) + owned_runs[route[0]] = owned_runs.get(route[0], 0) + 1 + age = run_age_minutes(run, clock) + owned_now[route[0]] = owned_now.get(route[0], 0) + young_charge(route[1], age) + if age is not None and age < LIVE_WINDOW_MINUTES: + live_now[route[0]] = live_now.get(route[0], 0) + young_charge(route[1], age) + live_runs[route[0]] = live_runs.get(route[0], 0) + 1 elif route == "ephemeral": ephemeral += 1 else: unknown += 1 - return Routed(unknown=unknown, owned=owned, ephemeral=ephemeral, owned_now=owned_now) + live_unknown += young(run) + return Routed(unknown=unknown, owned=owned, ephemeral=ephemeral, owned_now=owned_now, owned_runs=owned_runs, + live_now=live_now, live_runs=live_runs, live_unknown=live_unknown) def run_route(self, run: Mapping[str, Any]) -> tuple[str, int] | str | None: """(owned pool, peak), "ephemeral", or None while this run's pick is unknown.""" @@ -1857,7 +1961,11 @@ def summary(choice: Choice, snapshot: Mapping[str, Any] | None, *, now: dt.datet lines.append(f"- Compile admission takes {labels}: an idle root runner kept a build of this run's merge base") for problem in problems: lines.append(f"- **Error:** {problem}; that pool gets no machines") - if isinstance(snapshot, Mapping) and isinstance(snapshot.get("pools"), Mapping): + if isinstance(snapshot, Mapping) and snapshot.get("source") == "live": + lines.append("- No janitor snapshot: owned pools read live from the runners API:") + for label in sorted(owned_slots or {}): + lines.append(f" - {describe(snapshot, label, owned_slots)}") + elif isinstance(snapshot, Mapping) and isinstance(snapshot.get("pools"), Mapping): age = snapshot_age_minutes(snapshot, now) lines.append(f"- Queue seen by the janitor at {snapshot.get('generated_at')}" + (f" ({round(age)} min before this run)" if age is not None else "") + ":") diff --git a/tests/test_ci_pr_runner_pool.py b/tests/test_ci_pr_runner_pool.py index 7eb8726be9eb..e409d19274f0 100644 --- a/tests/test_ci_pr_runner_pool.py +++ b/tests/test_ci_pr_runner_pool.py @@ -602,7 +602,7 @@ def routed(since): choice = owned_choice(fleet(busy=0), live_owned={MINI: 5}, jobs=1, routed=routed) # 5 idle - 2 taken - 1 replayed run * REPLAYED_RUN_JOBS(3) = 0 < 1: Blacksmith. self.assertEqual(choice.runner, LARGE) - self.assertEqual(windows[1], (NOW - dt.timedelta(minutes=pool.LIVE_WINDOW_MINUTES)).strftime("%Y-%m-%dT%H:%M:%SZ")) + self.assertEqual(windows[1], (NOW - dt.timedelta(minutes=pool.DEFAULT_JOB_MINUTES)).strftime("%Y-%m-%dT%H:%M:%SZ")) windows.clear() self.assertEqual(owned_choice(fleet(busy=0), live_owned={MINI: 6}, jobs=1, routed=routed).runner, MINI) @@ -1321,22 +1321,25 @@ def test_a_seeded_queue_before_a_cold_free_machine(self): snap["pools"][SMALL]["queued"] = 10 self.assertEqual(choose(snap, queue_rounds="").runner, OLD) - def test_a_busy_fleet_queues_while_its_wait_is_under_blacksmiths(self): + def test_a_busy_fleet_queues_within_its_rounds_whatever_blacksmiths_wait(self): # Every mini busy; this run's 3 jobs would wait 9 minutes on 12vcpu, 25 on 6vcpu. busy = fleet(busy=11, small=21, large=6, old=4) self.assertEqual(owned_choice(busy, queue_rounds="0").runner, LARGE) choice = owned_choice(busy, queue_rounds="") - # 9 minutes over 11 minis of 10-minute jobs: 9 queue places. + # One round (10 minutes) over 11 minis of 10-minute jobs: 11 queue places. self.assertEqual((choice.runner, choice.owned_budget), (MINI, 3)) - self.assertIn("0 of 11 owned machines free and 9 queue places within 9 min " + self.assertIn("0 of 11 owned machines free and 11 queue places within 10 min " "(Blacksmith's expected wait 9 min)", choice.reason) - # 7 already queued leave 2 places, too few for 3 jobs. - fuller = fleet(busy=11, queued=7, small=21, large=6, old=4) + # 9 already queued leave 2 places, too few for 3 jobs. + fuller = fleet(busy=11, queued=9, small=21, large=6, old=4) self.assertEqual(owned_choice(fuller, queue_rounds="").runner, LARGE) split = owned_choice(fuller, queue_rounds="", split="1") self.assertEqual((split.runner, split.owned_budget), (MINI, 2)) - # A free Blacksmith machine means no wait: a busy fleet takes nothing. - self.assertEqual(owned_choice(fleet(busy=11, large=0), queue_rounds="").runner, LARGE) + # Fleet first: a free Blacksmith machine does not send the run off + # minis that free up within the round (Blacksmith is overflow). + idle_blacksmith = owned_choice(fleet(busy=11, large=0), queue_rounds="") + self.assertEqual((idle_blacksmith.runner, idle_blacksmith.owned_budget), (MINI, 3)) + self.assertIn("(Blacksmith's expected wait 0 min)", idle_blacksmith.reason) # And never more than CI_PR_POOL_QUEUE_ROUNDS job lengths, however long Blacksmith's queue. jammed = fleet(busy=11, queued=10, small=90, large=60, old=90) self.assertEqual(owned_choice(jammed, queue_rounds="").runner, LARGE) @@ -1369,16 +1372,17 @@ def test_root_runners_by_the_same_wait(self): self.assertIn("0 of 14 root runners free", old.reason) self.assertNotIn("-4 of", old.reason) choice = owned_choice(snap, queue_rounds="", **kwargs) - self.assertEqual((choice.runner, choice.root_runner, choice.root_budget), (MINI, ROOT_MINI, 4)) - # Newer runs younger than a job hold only admission and side lanes (3 of their 7). + self.assertEqual((choice.runner, choice.root_runner, choice.root_budget), (MINI, ROOT_MINI, 10)) + # Newer runs younger than a job hold only admission and side lanes (3 of + # their 7); the bound still counts their peaks. young = dict(kwargs, routed=pool.Routed(owned={MINI: 7}, owned_now={MINI: 3})) - self.assertEqual(owned_choice(snap, queue_rounds="", **young).root_budget, 8) - # Root runners all busy: they queue only within Blacksmith's wait for - # this run's 2 jobs (8 min: 11 places over 14 root runners). + self.assertEqual(owned_choice(snap, queue_rounds="", **young).root_budget, 10) + # Root runners all busy: they queue a round (14 places over 14 root + # runners), whatever Blacksmith's wait (8 min for this run's 2 jobs). snap = fleet(busy=14, small=21, large=6, old=4) snap["pools"][ROOT_MINI] = {"queued": 0, "running": 14} busy = owned_choice(snap, machines=36, owned_slots=slots, jobs=2, root_jobs=2, queue_rounds="") - self.assertEqual((busy.runner, busy.root_budget), (MINI, 11)) + self.assertEqual((busy.runner, busy.root_budget), (MINI, 14)) self.assertEqual(owned_choice(snap, machines=36, owned_slots=slots, jobs=2, root_jobs=2, queue_rounds="0").runner, LARGE) @@ -1420,45 +1424,52 @@ def test_a_marker_run_holds_its_admission_while_young_and_its_peak_after(self): self.assertEqual(routed.owned, {MINI: 22}) self.assertEqual(routed.owned_now, {MINI: pool.REPLAYED_RUN_JOBS + 11}) - def test_a_replayed_run_is_compared_at_the_jobs_it_has(self): - # 11 minis busy with 1 queued; 12vcpu full (5 running). Compared at one - # job, Blacksmith's wait is 1 minute: 1 queue place, none left, so the - # replayed run was put on 12vcpu and charged nothing on the busy minis. - # Compared at REPLAYED_RUN_JOBS (3 minutes, 3 places) it takes the minis. - snap = fleet(busy=11, queued=1, small=21, large=0, old=4) + def test_a_replayed_run_takes_the_busy_fleet_within_the_rounds(self): + # 11 minis busy with 6 queued; 12vcpu full (5 running) with a 1-minute + # wait. The replayed run takes the minis however it is compared with + # Blacksmith: 5 of the round's 11 places are left. + snap = fleet(busy=11, queued=6, small=21, large=0, old=4) snap["pools"][LARGE]["running"] = 5 load = {label: pool.pool(snap, label, {MINI: 11}) for label in (MINI, LARGE, SMALL)} added = {label: 0 for label in load} order = [MINI, LARGE, SMALL] - self.assertEqual(pool.pick(load, added, order, 0, jobs=1, queue_rounds=1).label, LARGE) + self.assertEqual(pool.pick(load, added, order, 0, jobs=1, queue_rounds=1).label, MINI) self.assertEqual(pool.pick(load, added, order, 0, jobs=1, queue_rounds=1, compared_jobs=pool.REPLAYED_RUN_JOBS).label, MINI) - # So the replay charges the minis, and this run finds them full. + # So the replay charges the minis REPLAYED_RUN_JOBS, and this run's 3 + # jobs find 2 places left. self.assertIn("after replaying 1 newer run", owned_choice(snap, routed=1, queue_rounds="").reason) self.assertEqual(owned_choice(snap, routed=1, queue_rounds="").runner, LARGE) - def test_live_busy_fleet_queues_within_blacksmiths_wait(self): + def test_live_busy_fleet_queues_within_its_rounds(self): busy = fleet(busy=0, small=21, large=6, old=4) - # Every runner busy, the API cannot see a queue: 9 places within 9 minutes. + # Every runner busy, the API cannot see a queue: a round is 11 places. choice = owned_choice(busy, live_owned={MINI: 0}, queue_rounds="") self.assertEqual((choice.runner, choice.owned_budget), (MINI, 3)) self.assertEqual(owned_choice(busy, live_owned={MINI: 0}, queue_rounds="0").runner, LARGE) - # Runs of the live window hold 7 jobs there: 2 places left. + # Runs of the live window hold 7 jobs there: 4 places left, then 2. recent = pool.Routed(owned={MINI: 7}) + self.assertEqual(owned_choice(busy, live_owned={MINI: 0}, queue_rounds="", routed=recent).runner, MINI) + recent = pool.Routed(owned={MINI: 9}) self.assertEqual(owned_choice(busy, live_owned={MINI: 0}, queue_rounds="", routed=recent).runner, LARGE) - # Older runs since the snapshot may still be queued where no runner is idle. + # Older runs since the snapshot may still have their admission queued + # where no runner is idle: one job per run, never their peaks. windows = [] def routed(since): windows.append(since) - return pool.Routed(owned={MINI: 7}) if len(windows) == 1 else pool.Routed() + return pool.Routed(owned={MINI: 7}, owned_runs={MINI: 7}) if len(windows) == 1 else pool.Routed() - self.assertEqual(owned_choice(busy, live_owned={MINI: 0}, queue_rounds="", routed=routed).runner, LARGE) + self.assertEqual(owned_choice(busy, live_owned={MINI: 0}, queue_rounds="", routed=routed).runner, MINI) + windows.clear() + self.assertEqual(owned_choice(busy, live_owned={MINI: 0}, queue_rounds="", jobs=5, routed=routed).runner, + LARGE) # An idle runner means that label has no queue: they are running. windows.clear() - self.assertEqual(owned_choice(busy, live_owned={MINI: 1}, queue_rounds="", routed=routed).runner, MINI) - # No Blacksmith wait, no queue. - self.assertEqual(owned_choice(fleet(busy=0), live_owned={MINI: 0}, queue_rounds="").runner, LARGE) + self.assertEqual(owned_choice(busy, live_owned={MINI: 1}, queue_rounds="", jobs=5, routed=routed).runner, + MINI) + # Fleet first: a free Blacksmith machine does not beat a round's queue. + self.assertEqual(owned_choice(fleet(busy=0), live_owned={MINI: 0}, queue_rounds="").runner, MINI) def test_forks_read_the_rounds_the_janitor_copied(self): snap = backlog(small=12, large=8, old=30, settings={**FORK_SETTINGS, "queue_rounds": "0"}) @@ -3094,5 +3105,127 @@ def test_home_secrets_are_removed_however_the_job_ends(self): self.assertTrue({"ios-streamed-validate.yml", "iroh-release-gate.yml"} <= checked, checked) +class LiveIdleRunners(unittest.TestCase): + """With the runners read live, idle runners decide and in-flight peaks are only the fallback.""" + + SLOTS = json.dumps({MINI: 40, ROOT_MINI: 19}) + + def test_live_roots_ignore_committed_peaks(self): + # The shape of run 36172350539 (2026-09-25): a full suite needing 9 root + # runners found all of them busy, and in-flight full suites' peaks + # (here 57 committed) filled the bound, so its root jobs went to Blacksmith. + snap = fleet(busy=40, small=0, large=0, old=0) + snap["pools"][MINI]["committed"] = 60 + snap["pools"][ROOT_MINI] = {"queued": 0, "running": 19, "committed": 57} + kwargs = dict(machines=40, owned_slots=self.SLOTS, jobs=10, root_jobs=9, queue_rounds="2", + split="1") + # Snapshot counts (no runners API): the committed peaks fill the bound. + fallback = owned_choice(snap, **kwargs) + self.assertEqual((fallback.runner, fallback.root_budget), (MINI, 0)) + # Live: 0 idle, but only what runs now holds the label, so a full + # suite's 9 root jobs queue within the rounds instead of Blacksmith. + live = owned_choice(snap, live_owned={MINI: 0, ROOT_MINI: 0}, live_online={MINI: 40, ROOT_MINI: 19}, + **kwargs) + self.assertEqual((live.runner, live.root_runner, live.owned_budget), (MINI, ROOT_MINI, 10)) + self.assertGreaterEqual(live.root_budget, 9) + self.assertIn("0 of 19 root runners free", live.reason) + + def test_a_failed_snapshot_download_leaves_the_owned_pools_to_the_live_runners(self): + windows = [] + + def routed(since): + windows.append(since) + return pool.Routed() + + def fail(): + raise RuntimeError("artifact download failed (503)") + + live = owned_choice(None, fetch=fail, live_owned={MINI: 6}, routed=routed) + self.assertEqual(live.runner, MINI) + self.assertIn("no janitor snapshot (could not read the pool snapshot (artifact download failed (503)))", + live.reason) + # Only the live window is counted: the snapshot's window does not exist. + self.assertEqual(windows, [pool.iso(NOW - dt.timedelta(minutes=pool.DEFAULT_JOB_MINUTES))]) + # Without the runners it keeps every job's default, as before. + self.assertEqual(owned_choice(None, fetch=fail).runner, "") + # A retry never reads the fleet off the live runners alone. + self.assertEqual(owned_choice(None, fetch=fail, live_owned={MINI: 6}, attempt=2).runner, "") + + def test_a_stale_or_missing_snapshot_does_not_skip_idle_minis(self): + for snap in (fleet(busy=11, age=120), None, {"generated_at": pool.iso(NOW)}): + live = owned_choice(snap, live_owned={MINI: 5}) + self.assertEqual(live.runner, MINI, snap) + self.assertIn("Blacksmith's queues are unknown", live.reason) + self.assertEqual(owned_choice(snap).runner, "", snap) + # Nothing idle and no queue known: the run queues a round on the fleet. + self.assertEqual(owned_choice(None, live_owned={MINI: 0}, queue_rounds="").runner, MINI) + # Past the round (11 machines, 11 places), overflow keeps every job's + # default: Blacksmith's queues are unknown, so no pool of it is picked. + full = owned_choice(None, live_owned={MINI: 0}, queue_rounds="", jobs=3, + routed=pool.Routed(owned={MINI: 20})) + self.assertEqual(full.runner, "") + self.assertIn("no readable pool snapshot; no owned pool has room", full.reason) + # A stale snapshot's warm keys are kept for warm affinity. + stale = fleet(busy=11, age=120) + stale["warm"] = {"runners": {"cmux1-glaeda": {"keys": ["pr-7"]}}} + _, snap = pool.choose( + event="pull_request", repo="manaflow-ai/cmux", head_repo="manaflow-ai/cmux", default_runner=SMALL, + overflow="", order="", max_queued="", xcode_pins=OWNED_PINS, owned="1", + owned_slots=json.dumps({MINI: 11}), jobs=3, fetch=lambda: stale, now=NOW, live_owned={MINI: 4}) + self.assertEqual((snap["source"], snap["warm"]), ("live", stale["warm"])) + + def test_runs_younger_than_a_job_bound_the_queue_by_their_peaks(self): + # A full suite that took the minis 5 minutes ago: its admission and + # side lanes run (3 of the 11 busy runners), its 8 later jobs are on + # the way. They count toward the bound, not toward the wait. + routed = pool.Routed(owned={MINI: 11}, owned_now={MINI: 3}, live_now={}, live_runs={}, live_unknown=0) + busy = fleet(busy=0) + kwargs = dict(live_owned={MINI: 0}, live_online={MINI: 11}, queue_rounds="", jobs=4, routed=routed) + # The fake returns the run for the snapshot's window too, so its + # admission may still be queued (older than the live window): 11 x + # (1 + 1) - 11 busy - 1 queued - 8 on the way = 2 places, too few for 4. + self.assertEqual(owned_choice(busy, **kwargs).runner, LARGE) + self.assertEqual(owned_choice(busy, **dict(kwargs, jobs=2)).runner, MINI) + # Without that run: a round of 11 places. + self.assertEqual(owned_choice(busy, **dict(kwargs, routed=pool.Routed())).runner, MINI) + # Runs past the live window whose route is unknown are not replayed: + # their jobs are on the runners already. + unknown = pool.Routed(unknown=4, live_now={}, live_runs={}, live_unknown=0) + self.assertEqual(owned_choice(busy, **dict(kwargs, routed=unknown)).runner, MINI) + self.assertEqual(owned_choice(busy, **dict(kwargs, routed=dataclasses.replace(unknown, live_unknown=3))).runner, + LARGE) + # A fork run still needs the janitor's copied settings. + self.assertEqual(choose(None, head="someone/cmux", default="", pins={}, live_owned={MINI: 9}).runner, "") + + def test_the_summary_says_the_owned_pools_were_read_live(self): + choice, snap = pool.choose( + event="pull_request", repo="manaflow-ai/cmux", head_repo="manaflow-ai/cmux", default_runner=SMALL, + overflow="", order="", max_queued="", xcode_pins=OWNED_PINS, owned="1", + owned_slots=json.dumps({MINI: 11}), jobs=3, fetch=lambda: None, now=NOW, live_owned={MINI: 4}) + text = pool.summary(choice, snap, now=NOW, owned_slots={MINI: 11}) + self.assertIn("No janitor snapshot: owned pools read live", text) + self.assertIn(f"{MINI}: 0 queued, 7 running of 11 slots", text) + self.assertNotIn("Queue seen by the janitor", text) + + def test_routes_count_the_runs_on_each_owned_pool(self): + client = pool.GitHub("t", "manaflow-ai/cmux") + repo = {"id": 1} + runs = [{"id": number, "run_attempt": 1, "status": "in_progress", "event": "pull_request", + "head_repository": repo, "repository": repo, + "created_at": pool.iso(NOW - dt.timedelta(minutes=minutes))} for number, minutes in ((1, 1), (2, 6))] + with unittest.mock.patch.object(client, "runs_since", return_value=runs), \ + unittest.mock.patch.object(client, "run_route", return_value=(MINI, 11)): + routed = client.pull_request_routes_since("x", exclude_run_id=None, now=NOW) + self.assertEqual((routed.owned, routed.runs()), ({MINI: 22}, {MINI: 2})) + # Only the run younger than LIVE_WINDOW_MINUTES holds machines toward the live wait. + self.assertEqual((routed.live_now, routed.live_runs), ({MINI: pool.REPLAYED_RUN_JOBS}, {MINI: 1})) + # An unreadable route is unknown, and counted in the live window only while young. + with unittest.mock.patch.object(client, "runs_since", return_value=runs), \ + unittest.mock.patch.object(client, "run_route", return_value=None): + routed = client.pull_request_routes_since("x", exclude_run_id=None, now=NOW) + self.assertEqual((routed.unknown, routed.live_unknown), (2, 1)) + self.assertEqual(pool.Routed(owned={MINI: 7, LIGHT: 0}).runs(), {MINI: 1}) + + if __name__ == "__main__": unittest.main(verbosity=2)