From 3468ed6c3df62544fffec974a4ce6b8386f7b748 Mon Sep 17 00:00:00 2001 From: Leo Li Date: Fri, 30 Oct 2026 15:26:01 +0000 Subject: [PATCH 1/3] test(ci): the janitor lists the whole queue, readers distrust a partial one, a handover keeps the cron's sweeper Red on main: the 1,138-run queue lists 300 runs, no snapshot says truncated, every pool reader trusts a partial snapshot, and a code handover dispatches a sweeper that cancels the one already waiting. Co-Authored-By: Claude Opus 5.5 Agent-Session: https://cmux.com/deeplink/agent?id=aid-0b4091cd6a9b72e19edf --- tests/test_ci_owned_pool_rescue.py | 42 +++++++++ tests/test_ci_pr_runner_pool.py | 43 +++++++++ tests/test_ci_queue_janitor.py | 136 ++++++++++++++++++++++++++++- tests/test_ci_runner_pools.py | 12 +++ 4 files changed, 232 insertions(+), 1 deletion(-) diff --git a/tests/test_ci_owned_pool_rescue.py b/tests/test_ci_owned_pool_rescue.py index f439a4bfd9f5..52e785bf311f 100644 --- a/tests/test_ci_owned_pool_rescue.py +++ b/tests/test_ci_owned_pool_rescue.py @@ -4301,6 +4301,48 @@ def test_hands_over_once_main_s_code_changes(self): self.assertEqual(api.dispatched, 1) self.assertEqual(watched, [1]) + def test_a_handover_dispatches_no_sweeper_when_one_already_waits(self): + # 2026-10-30: the handover's dispatch cancelled the cron's pending sweeper (run 41071107039) in the + # concurrency group before it ran. The waiting one checks out main, so it already runs the new code. + class Waiting(CodeSweepAPI): + def waiting_sweeper(self): + return 41071107039 + api = Waiting([listed(1)], picker=[1], changes_after=1) + _, _, outcomes = self.code_sweep(api) + self.assertEqual(outcomes.get("code changed"), 1) + self.assertEqual(api.dispatched, 0) + + class NoneWaiting(CodeSweepAPI): + def waiting_sweeper(self): + return None + + class Unreadable(CodeSweepAPI): + def waiting_sweeper(self): + raise urllib.error.URLError("down") + for cls in (NoneWaiting, Unreadable): # unknown dispatches as before, so a sweeper surely follows + with self.subTest(api=cls.__name__): + api = cls([listed(1)], picker=[1], changes_after=1) + self.code_sweep(api) + self.assertEqual(api.dispatched, 1) + + def test_waiting_sweeper_finds_a_queued_or_pending_sweeper_but_not_itself_or_a_rescue(self): + client = rescue.GitHub("token", "manaflow-ai/cmux") + client.own_run = 5 + listings = {"pending": [{"id": 5, "display_title": "owned-pool-sweeper"}, + {"id": 6, "display_title": "owned-pool-rescue-77-1"}], + "queued": [{"id": 8, "display_title": "owned-pool-sweeper"}]} + asked = [] + + def request(method, path, **kwargs): + asked.append(path) + status = path.split("status=", 1)[1].split("&", 1)[0] + return {"workflow_runs": listings.get(status, [])} + client.request = request + self.assertEqual(client.waiting_sweeper(), 8) + self.assertEqual([path.split("?")[1].split("&")[0] for path in asked], ["status=pending", "status=queued"]) + listings["queued"] = [] + self.assertIsNone(client.waiting_sweeper()) + def test_unchanged_or_unreadable_code_keeps_sweeping(self): for api in (CodeSweepAPI([], changes_after=None), CodeSweepAPI([], fails=True)): clock, _, outcomes = self.code_sweep(api, ticks=12) diff --git a/tests/test_ci_pr_runner_pool.py b/tests/test_ci_pr_runner_pool.py index 84442e1abc00..f44f21faad08 100644 --- a/tests/test_ci_pr_runner_pool.py +++ b/tests/test_ci_pr_runner_pool.py @@ -4268,3 +4268,46 @@ def test_routes_count_the_runs_on_each_owned_pool(self): if __name__ == "__main__": unittest.main(verbosity=2) + + +class TruncatedSnapshotTests(unittest.TestCase): + """A janitor sweep that listed part of the queue (queue_janitor.listing_state()) is no view of it: + on 2026-10-30 a sweep listed 300 of 1,138 queued runs, and its pools read the rest as nothing queued.""" + + def test_a_truncated_snapshot_is_a_problem_and_a_complete_one_is_not(self): + fresh = backlog() + self.assertEqual(pool.snapshot_problem(fresh, NOW), "") # written before the key existed + self.assertEqual(pool.snapshot_problem(dict(fresh, truncated=False, listed=9, total=9), NOW), "") + for value in (True, 1, "yes", None): + with self.subTest(truncated=value): + problem = pool.snapshot_problem(dict(fresh, truncated=value, listed=300, total=1138), NOW) + self.assertIn("truncated", problem) + self.assertIn("300 of 1138", pool.snapshot_truncated(dict(fresh, truncated=True, listed=300, total=1138))) + + def test_decide_keeps_the_default_on_a_truncated_snapshot(self): + # An empty large pool would take the run (test_placed_runs_and_a_narrower_pick_for_e2e); truncated, it + # may hold the demand the sweep never listed, so the default keeps its say. + snap = backlog(small=0, large=0, old=0) + snap["pools"][LARGE]["running"] = 0 + args = dict(now=NOW, xcode_pins=PINS) + self.assertEqual(pool.decide(snap, pool.Settings(), **args).runner, LARGE) + choice = pool.decide(dict(snap, truncated=True, listed=300, total=1138), pool.Settings(), **args) + self.assertEqual(choice.runner, "") + self.assertIn("truncated", choice.reason) + + def test_the_e2e_and_ios_load_reads_a_truncated_snapshot_as_none(self): + class Client: + def __init__(self, snapshot): + self.snap = snapshot + + def snapshot(self, *, now): + return self.snap + + def runs_since(self, workflow, since, **filters): + return [] + + def pull_request_runs_since(self, since, *, exclude_run_id): + return 0 + + self.assertIsNotNone(e2e_pool.measure_load(Client(backlog()), now=NOW, exclude_run_id=None)) + self.assertIsNone(e2e_pool.measure_load(Client(dict(backlog(), truncated=True)), now=NOW, exclude_run_id=None)) diff --git a/tests/test_ci_queue_janitor.py b/tests/test_ci_queue_janitor.py index 6c674a656bbd..1eee684d5bea 100644 --- a/tests/test_ci_queue_janitor.py +++ b/tests/test_ci_queue_janitor.py @@ -3454,7 +3454,141 @@ def test_only_an_empty_page_or_the_page_cap_ends_a_status(self): self.assertEqual([page for status, page in asked if status == "queued"], [1, 2]) pages = {page: [{"id": page * 1000 + index} for index in range(100)] for page in range(1, 5)} ids, asked = self.listing({"queued": pages}) - self.assertEqual(len(ids), 100 * janitor.MAX_RUN_PAGES) + self.assertEqual(len(ids), 400) # four full pages, then the empty fifth: no cap at three + + +def github_listing(runs, *, result_pages=10): + """A janitor client whose run listing answers like GitHub: newest first, 100 a page, `status` and + `created<=` filters, and nothing past `result_pages` pages (its 1,000-result cap). Returns (client, asked).""" + github = janitor.GitHub("token", "manaflow-ai/cmux") + asked = [] + ordered = sorted(runs, key=lambda run: (run["created_at"], run["id"]), reverse=True) + + def request(method, path, body=None): + query = urllib.parse.parse_qs(path.split("?", 1)[1]) + status, page = query["status"][0], int(query["page"][0]) + created = query.get("created", [""])[0] + asked.append((status, page, created)) + rows = [run for run in ordered if run["status"] == status + and (not created or run["created_at"] <= created.removeprefix("<="))] + batch = rows[(page - 1) * 100:page * 100] if page <= result_pages else [] + return {"workflow_runs": batch, "total_count": len(rows)} + + github.request = request + return github, asked + + +def queued_runs(count, *, status="queued"): + """`count` runs a second apart, newest first in time.""" + start = dt.datetime(2026, 10, 30, 14, 0, tzinfo=dt.timezone.utc) + return [{"id": 10_000 + index, "status": status, + "created_at": (start - dt.timedelta(seconds=index)).strftime("%Y-%m-%dT%H:%M:%SZ")} + for index in range(count)] + + +class FullQueueListingTests(unittest.TestCase): + """2026-10-30: 1,138 runs queued, and the sweep listed 300 of them (three pages a status).""" + + def test_every_queued_run_of_a_1138_run_queue_is_listed(self): + runs = queued_runs(1138) + github, asked = github_listing(runs) + ids = {run["id"] for run in github.in_flight_runs()} + self.assertEqual(ids, {run["id"] for run in runs}) + state = janitor.listing_state(github, set()) + self.assertEqual((state["truncated"], state["listed"], state["total"]), (False, 1138, 1138)) + self.assertEqual(state["statuses"]["queued"]["listed"], 1138) + # Past GitHub's ten pages the listing goes on from the oldest run seen, not from page eleven. + self.assertNotIn(("queued", 11, ""), asked) + self.assertTrue(any(created.startswith("<=") for status, _, created in asked if status == "queued")) + + def test_a_listing_past_its_ceiling_says_truncated_with_counts(self): + runs = queued_runs(500) + github, asked = github_listing(runs) + with mock.patch.object(janitor, "MAX_RUN_PAGES", 2): + ids = github.in_flight_runs() + self.assertEqual(len(ids), 200) + self.assertEqual(len([entry for entry in asked if entry[0] == "queued"]), 2) + state = janitor.listing_state(github, set()) + self.assertIs(state["truncated"], True) + self.assertEqual((state["listed"], state["total"]), (200, 500)) + self.assertIs(state["statuses"]["queued"]["truncated"], True) + + def test_runs_left_past_the_job_listing_cap_truncate_the_listing(self): + runs = [make_run(status="queued", age=age, branch=f"b{age}") for age in (5, 10, 15)] + fake = FakeGitHub({run["id"]: dict(run) for run in runs}, {run["id"]: mac_jobs(queued=1) for run in runs}) + jobs, unread, unlisted = janitor.read_run_jobs(fake, runs, frozenset(), NOW, cap=2) + # The newest two are listed; the oldest is left, and the snapshot says so. + self.assertEqual(set(jobs), {runs[0]["id"], runs[1]["id"]}) + self.assertEqual((unread, unlisted), (set(), {runs[2]["id"]})) + state = janitor.listing_state(fake, unlisted) + self.assertIs(state["truncated"], True) + self.assertEqual(state["jobs_unlisted"], 1) + + def test_parallel_job_listings_match_one_at_a_time(self): + runs = [make_run(status="queued", age=age, branch=f"b{age}") for age in range(1, 40)] + job_map = {run["id"]: mac_jobs(queued=run["id"] % 3 + 1) for run in runs} + fake = FakeGitHub({run["id"]: dict(run) for run in runs}, job_map) + fake.unreadable_jobs = {runs[7]["id"]} + with mock.patch.object(janitor.time, "sleep", lambda _: None), mock.patch("builtins.print"): + parallel = janitor.read_run_jobs(fake, runs, frozenset(), NOW, workers=6) + serial = janitor.read_run_jobs(fake, runs, frozenset(), NOW, workers=1) + self.assertEqual(parallel, serial) + self.assertEqual(list(parallel[0]), [run["id"] for run in runs if run["id"] != runs[7]["id"]]) + self.assertEqual(parallel[1], {runs[7]["id"]}) + + def test_a_rate_limit_in_a_parallel_listing_ends_the_sweep(self): + runs = [make_run(status="queued", age=age, branch=f"b{age}") for age in range(1, 10)] + + class Limited(FakeGitHub): + def jobs(self, run_id): + if run_id == runs[4]["id"]: + raise janitor.RateLimited("GET /jobs failed (403: rate limit)") + return [] + + with self.assertRaises(janitor.RateLimited): + janitor.read_run_jobs(Limited({}, {}), runs, frozenset(), NOW, workers=4) + + +class TruncatedSweepTests(unittest.TestCase): + """A sweep that saw part of the queue says so in the snapshot and decides nothing.""" + + def sweep(self, **patches): + stuck = make_run(status="queued", age=340, branch="ci/macos26-app-host-repair") + busy = make_run(status="in_progress", age=20, branch="busy") + runs = {r["id"]: dict(r) for r in (stuck, busy)} + jobs = {stuck["id"]: [linux_job(age=330), orphan_job(age=220)], busy["id"]: [served_job(age=15)]} + fake = FakeGitHub(runs, jobs) + with tempfile.TemporaryDirectory() as temp: + load, summary = Path(temp) / "load.json", Path(temp) / "summary.md" + argv = ["--repo", "manaflow-ai/cmux", "--summary", str(summary), "--pool-load", str(load), + "--workflows-dir", str(Path(temp) / "none")] + with mock.patch.object(janitor, "GitHub", lambda token, repo, fallback_token="": fake), \ + mock.patch.dict("os.environ", {"GH_TOKEN": "t"}), \ + mock.patch.object(janitor.time, "sleep", lambda _: None), \ + mock.patch.object(janitor, "utc_now", lambda: NOW), \ + mock.patch.multiple(janitor, MAX_JOB_LISTINGS=patches.get("MAX_JOB_LISTINGS", janitor.MAX_JOB_LISTINGS)), \ + mock.patch("builtins.print"): + code = janitor.main(argv) + return code, fake, json.loads(load.read_text()), summary.read_text() + + def test_a_complete_sweep_says_so_and_acts(self): + code, fake, snapshot, text = self.sweep() + self.assertEqual(code, 0) + self.assertIs(snapshot["truncated"], False) + self.assertEqual(snapshot["listed"], 2) + self.assertIn("queued", snapshot["listing"]["statuses"]) + self.assertNotIn("starved_incomplete", snapshot) + self.assertTrue(fake.posts()) # the orphan is cancelled + self.assertIn("Listed 2 of", text) + + def test_a_truncated_sweep_marks_the_snapshot_and_cancels_nothing(self): + code, fake, snapshot, text = self.sweep(MAX_JOB_LISTINGS=1) + self.assertEqual(code, 0) + self.assertIs(snapshot["truncated"], True) + self.assertEqual(snapshot["listing"]["jobs_unlisted"], 1) + self.assertIs(snapshot["starved_incomplete"], True) + self.assertEqual(fake.posts(), []) + self.assertIn("this sweep decided nothing", text) class OpenHeadNeverCancelledTests(unittest.TestCase): diff --git a/tests/test_ci_runner_pools.py b/tests/test_ci_runner_pools.py index eb1e4dfb7301..70f37cfdb53a 100644 --- a/tests/test_ci_runner_pools.py +++ b/tests/test_ci_runner_pools.py @@ -298,3 +298,15 @@ def test_check_fails_on_an_unset_or_malformed_variable(self) -> None: if __name__ == "__main__": unittest.main() + + +class TruncatedSnapshot(unittest.TestCase): + def test_a_truncated_snapshot_is_no_snapshot_evidence(self) -> None: + # A janitor sweep that listed part of the queue (queue_janitor.listing_state()) undercounts every pool. + stalled = copy.deepcopy(RECORDED) + stalled["pools"]["blacksmith-6vcpu-macos-26"].update(queued=3, oldest_queued_minutes=30) + dead, _, _ = runner_pools.label_evidence(None, stalled, now=NOW) + self.assertEqual(dead, {"blacksmith-6vcpu-macos-26"}) + dead, started, notes = runner_pools.label_evidence(None, dict(stalled, truncated=True), now=NOW) + self.assertEqual((dead, started), (set(), set())) + self.assertTrue(any("truncated" in note for note in notes)) From cf794e5e415487ba15abb37329c5a69b10371ddb Mon Sep 17 00:00:00 2001 From: Leo Li Date: Fri, 30 Oct 2026 15:27:50 +0000 Subject: [PATCH 2/3] ci(janitor): list every queued run, say when a listing is partial, keep the cron's sweeper The queue janitor listed at most three pages a status: with 1,138 runs queued on 2026-10-30 it saw 300, and the macos-pool-load snapshot read every other run as nothing queued. It now pages each status until an empty page, past GitHub's 1,000-result cap in `created<=` windows, up to 40 pages a status, and lists the jobs six at a time (at most 2,000 a sweep). The snapshot carries `truncated`, `listed`, `total` and a per-status `listing`; a truncated sweep decides nothing and asserts no starved label, and every pool reader (snapshot_problem(), decide(), the E2E/iOS load, the runner-pool evidence) takes a truncated snapshot as unknown, never as nothing queued. The rescue sweeper's code handover dispatched a new sweeper even when the cron's was already pending, and the concurrency group cancelled the pending one before it ran (run 41071107039). A handover now dispatches only when no sweeper waits. Co-Authored-By: Claude Opus 5.5 Agent-Session: https://cmux.com/deeplink/agent?id=aid-0b4091cd6a9b72e19edf --- .github/workflows/ci-queue-janitor.yml | 7 +- scripts/ci/e2e_runner_pool.py | 2 + scripts/ci/owned_pool_rescue.py | 31 ++++- scripts/ci/pr_runner_pool.py | 17 ++- scripts/ci/queue_janitor.py | 183 +++++++++++++++++++++---- scripts/ci/runner_pools.py | 2 + 6 files changed, 209 insertions(+), 33 deletions(-) diff --git a/.github/workflows/ci-queue-janitor.yml b/.github/workflows/ci-queue-janitor.yml index ef572cd08ce3..f1c0b359c79a 100644 --- a/.github/workflows/ci-queue-janitor.yml +++ b/.github/workflows/ci-queue-janitor.yml @@ -56,8 +56,11 @@ name: CI queue janitor # skips when the newest snapshot is under 600 s old, one cron interval, so it # only stands in for a late cron. A cron sweep skips when a snapshot is under # 300 s old, the case where a requested CI run has just stood in for it. A -# full sweep costs 130 to 170 API requests from the GITHUB_TOKEN budget every -# workflow in the repository shares. With a 240 s skip for requested runs and +# full sweep costs about one API request per in-flight run (its job listing) +# plus one per 100 runs listed: about 1,170 for the 1,160 runs in flight at +# 15:30Z on 2026-10-30, at most MAX_JOB_LISTINGS (2,000) job listings a sweep +# (scripts/ci/queue_janitor.py). In 2026-10 that was 130 to 170 requests from +# the GITHUB_TOKEN budget every workflow in the repository shares. With a 240 s skip for requested runs and # none for the cron, CI being requested hundreds of times an hour drove about # 12 sweeps an hour, and the budget ran out on 2026-10-06. Now there are about # 6. The concurrency group keeps one sweep running and one waiting. diff --git a/scripts/ci/e2e_runner_pool.py b/scripts/ci/e2e_runner_pool.py index ad9c5fd0dda3..f1b5beb9308d 100644 --- a/scripts/ci/e2e_runner_pool.py +++ b/scripts/ci/e2e_runner_pool.py @@ -306,6 +306,8 @@ def measure_load(client: ApiClient, *, now: dt.datetime, exclude_run_id: int | N snapshot = client.snapshot(now=now) if not isinstance(snapshot, Mapping) or not snapshot.get("generated_at"): return None + if pr_runner_pool.snapshot_truncated(snapshot): + return None # a partial queue is no usable snapshot (the default keeps its say) since = str(snapshot["generated_at"]) load = PoolLoad( snapshot, diff --git a/scripts/ci/owned_pool_rescue.py b/scripts/ci/owned_pool_rescue.py index 89f2b614b339..6909d19ddda5 100644 --- a/scripts/ci/owned_pool_rescue.py +++ b/scripts/ci/owned_pool_rescue.py @@ -2100,6 +2100,17 @@ def code_shas(self, names: Sequence[str]) -> dict[str, str]: if isinstance(item, Mapping)} return {name: found.get(name, "") for name in names} + def waiting_sweeper(self) -> int | None: + """The id of a sweeper run other than this one that has not started yet (queued, or pending in the + sweeper concurrency group), else None. One listing per status, made only at a handover.""" + for status in SWEEPER_WAITING_STATUSES: + data = self.request("GET", f"/actions/workflows/{RESCUE_WORKFLOW_FILE}/runs?status={status}&per_page=100") + for run in (data or {}).get("workflow_runs") or []: + if (isinstance(run, Mapping) and str(run.get("display_title") or "") == SWEEPER_TITLE + and int(run.get("id") or 0) != self.own_run): + return int(run["id"]) + return None + def dispatch_sweeper(self) -> None: """Start a sweeper on MAIN_BRANCH's code. It queues in the sweeper concurrency group until this one exits.""" self.request("POST", f"/actions/workflows/{RESCUE_WORKFLOW_FILE}/" + "dispatches", own_token=True, @@ -4383,6 +4394,13 @@ def sweep_target(run: Mapping[str, Any], repository: str, *, late: bool, full_re "ui_tests_dispatch.py") CODE_CHECK_TICKS = 5 RESCUE_WORKFLOW_FILE = "ci-owned-pool-rescue.yml" +# A handover dispatches a sweeper only when none waits already (GitHub.waiting_sweeper()). The sweeper +# concurrency group holds one running and one pending run, and a newer pending run cancels the older one: +# the handover's dispatch cancelled the cron's pending sweeper before it ran (run 41071107039, cancelled +# at 14:56:46Z on 2026-10-30 by the dispatch of run 41076139116 three seconds earlier; 41055530432 at +# 14:08Z the same way). The waiting one checks out main when it starts, so it already runs the new code. +SWEEPER_TITLE = "owned-pool-sweeper" +SWEEPER_WAITING_STATUSES = ("pending", "queued") def blob_sha(data: bytes) -> str: @@ -4596,9 +4614,18 @@ def adopt(run_id: int, late: bool, *, refusals_known: bool = True) -> bool: "new code") with lock: outcomes["code changed"] = outcomes.get("code changed", 0) + 1 + waiting = None + try: + waiting = client.waiting_sweeper() if hasattr(client, "waiting_sweeper") else None + except Exception as error: # noqa: BLE001 - unknown: dispatch as before, so a sweeper surely follows + log(f"could not list waiting sweepers ({type(error).__name__}: {error}); dispatching one") try: - # It waits in the sweeper concurrency group (never cancel-in-progress) until this one exits. - client.dispatch_sweeper() + if waiting: + # A second dispatch would cancel it in the concurrency group before it ran. + log(f"sweeper run {waiting} already waits to start; it runs main's new code, so none is dispatched") + else: + # It waits in the sweeper concurrency group (never cancel-in-progress) until this one exits. + client.dispatch_sweeper() except Exception as error: # noqa: BLE001 - the cron's next sweeper starts on the new code anyway log(f"could not start a sweeper ({type(error).__name__}: {error}); the cron's next one takes over") log("handing over: stopping watches; rescues under way finish") diff --git a/scripts/ci/pr_runner_pool.py b/scripts/ci/pr_runner_pool.py index a21347207136..566537ca7dc7 100644 --- a/scripts/ci/pr_runner_pool.py +++ b/scripts/ci/pr_runner_pool.py @@ -1238,8 +1238,19 @@ def owned_room(label: str, counts: Mapping[str, int], added_jobs: int, taken_pea return min(wait, bound) +def snapshot_truncated(snapshot: Mapping[str, Any]) -> str: + """Why `snapshot`'s queue counts are partial, or "": the janitor marks a sweep that could not list every + in-flight run `truncated` (queue_janitor.listing_state()). Its pools then miss demand, so a reader takes + the queue as unknown, never as nothing queued. Any value but false counts as truncated; a snapshot + without the key (written before it existed) does not.""" + if "truncated" not in snapshot or snapshot.get("truncated") is False: + return "" + return (f"pool snapshot is truncated (it lists {snapshot.get('listed', '?')} of {snapshot.get('total', '?')} " + "in-flight runs), so its queues are unknown") + + def snapshot_problem(snapshot: Any, now: dt.datetime) -> str: - """Why decide() would not read `snapshot` (missing, undated, stale or without pools), or "".""" + """Why decide() would not read `snapshot` (missing, undated, stale, truncated 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): @@ -1247,7 +1258,7 @@ def snapshot_problem(snapshot: Any, now: dt.datetime) -> str: 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 "" + return snapshot_truncated(snapshot) def iso(moment: dt.datetime) -> str: @@ -1690,6 +1701,8 @@ def decide( age = snapshot_age_minutes(snapshot, now) if age is None or age < -5 or age > MAX_SNAPSHOT_MINUTES: return Choice("", "", f"pool snapshot is stale or undated (age {age if age is None else round(age)} min)") + if snapshot_truncated(snapshot): + return Choice("", "", snapshot_truncated(snapshot)) try: load = {label: pool(snapshot, label, owned_slots) for label in limits.order} roots = {label: pool(snapshot, root_label(label), owned_slots) for label in limits.order diff --git a/scripts/ci/queue_janitor.py b/scripts/ci/queue_janitor.py index dcb197b5b7b4..12f264e61bbd 100644 --- a/scripts/ci/queue_janitor.py +++ b/scripts/ci/queue_janitor.py @@ -185,6 +185,7 @@ import os import re import sys +import threading import time import urllib.error import urllib.parse @@ -192,6 +193,7 @@ import zipfile import zlib from collections.abc import Iterable, Mapping, Sequence +from concurrent.futures import ThreadPoolExecutor from pathlib import Path from typing import Any @@ -215,8 +217,20 @@ API = "https://api.github.com" DEFAULT_THRESHOLD = 6 DEFAULT_MAX_CANCELS = 10 -MAX_RUN_PAGES = 3 +# The in-flight run listing pages each status until an empty page, up to MAX_RUN_PAGES pages a status +# (4,000 runs). Until 2026-10-30 it stopped at 3 pages: with 1,138 runs queued that day the sweep saw +# 300 of them, and the pool snapshot read every other one's jobs as nothing queued. A filtered +# listing answers at most LISTING_RESULT_PAGES pages (GitHub's 1,000-result cap), so a status that +# fills them is listed again from the oldest run seen (`created<=`), window by window. A listing +# that hits MAX_RUN_PAGES is marked truncated in the snapshot (listing_state()). +MAX_RUN_PAGES = 40 +LISTING_RESULT_PAGES = 10 MAX_JOB_PAGES = 3 +# Each listed run's jobs are one request; at most MAX_JOB_LISTINGS a sweep (about twice the 1,100 the +# 2026-10-30 queue needed), JOB_LISTING_WORKERS at a time: one at a time took 0.5 s a request, and +# 1,100 of them would not fit the workflow's 10 minutes. Runs past the cap leave the listing truncated. +MAX_JOB_LISTINGS = 2000 +JOB_LISTING_WORKERS = 6 GRAPHQL_BATCH = 25 # A read that meets a 5xx or a dropped connection is made again after each of these delays (seconds), # then fails. At 08:08Z on 2026-10-26 one 502 on a run's job listing ended the sweep before it wrote @@ -2977,6 +2991,10 @@ def __init__(self, token: str, repo: str, fallback_token: str = "") -> None: self._token = token self.calls = 0 self.kinds: dict[str, int] = {} # requests by request_kind(), for the summary's REST line + # The job listings run on JOB_LISTING_WORKERS threads (read_run_jobs()); the counters are shared. + self._count_lock = threading.Lock() + # in_flight_runs()'s account of its listing, per status (listing_state()). + self.run_listing: dict[str, dict[str, Any]] = {} self.fallback_token = fallback_token if fallback_token != token else "" # An App's token is in use: one was minted beside the workflow's, and it was not refused yet. self.app_token = bool(self.fallback_token) @@ -3005,9 +3023,10 @@ def request(self, method: str, path: str, body: Any | None = None, *, app_only: raise AssertionError("unreachable") def _request(self, method: str, path: str, body: Any | None = None, *, app_only: bool = False) -> Any: - self.calls += 1 kind = request_kind(method, path, self.repo) - self.kinds[kind] = self.kinds.get(kind, 0) + 1 + with self._count_lock: + self.calls += 1 + self.kinds[kind] = self.kinds.get(kind, 0) + 1 data = json.dumps(body).encode() if body is not None else None headers = dict(self.headers) if data is not None: @@ -3048,26 +3067,63 @@ def note_rate_limit(self, headers: Any) -> None: self.rate_limits[found[0]] = found[1] def in_flight_runs(self) -> list[dict[str, Any]]: - """Every run in IN_FLIGHT_RUN_STATUSES, up to MAX_RUN_PAGES pages a status. + """Every run in IN_FLIGHT_RUN_STATUSES, up to MAX_RUN_PAGES pages a status; how complete each status's + listing is goes to `run_listing` (listing_state()). A status-filtered listing can return a short page before its last one: `status=queued` answered 99 runs on page 1 and 85 on page 2 of a total_count of 182 at 06:10Z on 2026-10-26. Stopping at the first short page dropped every queued run older than page 1, so the snapshot's oldest wait on blacksmith-12vcpu-macos-26 read 24 minutes while run 40259152940's fast tier had waited 205, and the janitor never cancelled that run after PR #24021 landed. So only an - empty page ends a status before MAX_RUN_PAGES: neither a short page nor total_count, which a page + empty page ends a status: neither a short page nor total_count, which a page repeating rows or a listing that shifts between pages would satisfy early. Runs keyed by id drop the repeats; a run a shift moves past a page boundary is read on the next sweep. + + GitHub answers a filtered listing's first 1,000 results only (LISTING_RESULT_PAGES pages): a status + whose last such page still had runs is listed again from the oldest run it showed (`created<=` that + time, newest first), until a window ends in an empty page. A status that reaches MAX_RUN_PAGES, or a + window that cannot move past its oldest run, is truncated. """ runs: dict[int, dict[str, Any]] = {} + self.run_listing = {} for status in IN_FLIGHT_RUN_STATUSES: - for page in range(1, MAX_RUN_PAGES + 1): - query = urllib.parse.urlencode({"status": status, "per_page": 100, "page": page}) - payload = self.request("GET", f"/repos/{self.repo}/actions/runs?{query}") - batch = payload.get("workflow_runs") or [] - for run in batch: - runs[run["id"]] = run - if not batch: + seen: set[int] = set() + total: int | None = None + pages = 0 + before = "" # the window's `created<=` bound; "" for the first window + truncated = False + while True: + oldest = "" + ended = False + for page in range(1, LISTING_RESULT_PAGES + 1): + if pages >= MAX_RUN_PAGES: + truncated = True + break + params = {"status": status, "per_page": 100, "page": page} + if before: + params["created"] = f"<={before}" + payload = self.request("GET", f"/repos/{self.repo}/actions/runs?{urllib.parse.urlencode(params)}") + pages += 1 + if total is None and isinstance(payload.get("total_count"), int): + total = payload["total_count"] + batch = payload.get("workflow_runs") or [] + if not batch: + ended = True + break + for run in batch: + runs[run["id"]] = run + seen.add(run["id"]) + created = str(run.get("created_at") or "") + if created and (not oldest or created < oldest): + oldest = created + if ended or truncated: break + # Every page of the window had runs: the 1,000-result cap may hide older ones. + if not oldest or oldest == before: + truncated = True # no older bound to move to (over 1,000 runs created in one second) + break + before = oldest + self.run_listing[status] = {"listed": len(seen), "total": total if total is not None else len(seen), + "pages": pages, "truncated": truncated} return list(runs.values()) def artifact_names(self, run_id: int, *, stop: str) -> list[str]: @@ -3253,24 +3309,83 @@ class _Transient(Exception): def read_run_jobs(github: GitHub, runs: Sequence[Mapping[str, Any]], linux_only: frozenset[str], - now: dt.datetime) -> tuple[dict[int, list[dict[str, Any]]], set[int]]: - """({run id: jobs} for the runs needs_jobs() lists, the ids whose listing failed). + now: dt.datetime, *, cap: int | None = None, + workers: int | None = None) -> tuple[dict[int, list[dict[str, Any]]], set[int], set[int]]: + """({run id: jobs} for the runs needs_jobs() lists, the ids whose listing failed, the ids left unlisted). One run's failed listing skips that run, not the sweep (a rate limit still ends it): the sweep - that ended on one 502 at 08:08Z on 2026-10-26 wrote no pool snapshot (RETRY_DELAYS).""" + that ended on one 502 at 08:08Z on 2026-10-26 wrote no pool snapshot (RETRY_DELAYS). + At most `cap` (MAX_JOB_LISTINGS) runs are listed, the newest first; the rest are left unlisted and + the snapshot says it is truncated. The listings run `workers` (JOB_LISTING_WORKERS) at a time.""" + cap = MAX_JOB_LISTINGS if cap is None else cap + workers = JOB_LISTING_WORKERS if workers is None else workers + wanted = [run for run in runs if needs_jobs(run, linux_only, now)] + # Newest first, so a capped sweep still sees the demand that is arriving now. + wanted.sort(key=lambda run: (str(run.get("created_at") or ""), run["id"]), reverse=True) + listed, unlisted = wanted[:max(0, cap)], {run["id"] for run in wanted[max(0, cap):]} jobs_by_run: dict[int, list[dict[str, Any]]] = {} unread: set[int] = set() - for run in runs: - if not needs_jobs(run, linux_only, now): - continue + + def one(run: Mapping[str, Any]) -> list[dict[str, Any]] | None: try: - jobs_by_run[run["id"]] = github.jobs(run["id"]) + return github.jobs(run["id"]) except RateLimited: raise except RuntimeError as error: - unread.add(run["id"]) print(f"queue-janitor: run {run['id']} skipped this sweep: {error}", file=sys.stderr) - return jobs_by_run, unread + return None + + if workers <= 1 or len(listed) <= 1: + results = [one(run) for run in listed] + else: + with ThreadPoolExecutor(max_workers=workers) as pool: + futures = [pool.submit(one, run) for run in listed] + try: + results = [future.result() for future in futures] + except RateLimited: + for future in futures: + future.cancel() + raise + # In run order, whatever order the listings finished in. + by_id = {run["id"]: jobs for run, jobs in zip(listed, results)} + for run in runs: + if run["id"] not in by_id: + continue + if by_id[run["id"]] is None: + unread.add(run["id"]) + else: + jobs_by_run[run["id"]] = by_id[run["id"]] + return jobs_by_run, unread, unlisted + + +def listing_state(github: Any, unlisted: set[int]) -> dict[str, Any]: + """The snapshot's account of how much of the in-flight queue this sweep saw. + + `listed` runs of GitHub's `total` (its total_count per status, as the first page reported it; the + queue moves while the pages are read, so the two can differ by a few either way), per status in + `statuses`. `truncated` is true when a status's listing hit MAX_RUN_PAGES or runs were left past + MAX_JOB_LISTINGS (`jobs_unlisted`): the snapshot's pools then miss demand, and its readers take the + queue as unknown (pr_runner_pool.snapshot_truncated()), never as nothing queued.""" + statuses = {status: dict(entry) for status, entry in (getattr(github, "run_listing", None) or {}).items()} + truncated = any(entry.get("truncated") for entry in statuses.values()) or bool(unlisted) + return { + "truncated": truncated, + "listed": sum(int(entry.get("listed") or 0) for entry in statuses.values()), + "total": sum(int(entry.get("total") or 0) for entry in statuses.values()), + "jobs_unlisted": len(unlisted), + "statuses": statuses, + } + + +def render_listing(listing: Mapping[str, Any]) -> str: + """`Listed 866 of 866 in-flight runs (queued 600/600, ...)`, plus what was cut when truncated.""" + parts = ", ".join(f"{status} {entry.get('listed', 0)}/{entry.get('total', 0)}" + + (" truncated" if entry.get("truncated") else "") + for status, entry in (listing.get("statuses") or {}).items()) + text = f"Listed {listing.get('listed', 0)} of {listing.get('total', 0)} in-flight runs ({parts or 'none'})" + if listing.get("jobs_unlisted"): + text += f"; {listing['jobs_unlisted']} run(s) past the {MAX_JOB_LISTINGS} job listings a sweep left unread" + return text def main(argv: Sequence[str] | None = None) -> int: @@ -3324,7 +3439,8 @@ def main(argv: Sequence[str] | None = None) -> int: linux_only = linux_only_workflow_paths(args.workflows_dir) try: runs = github.in_flight_runs() - jobs_by_run, unread = read_run_jobs(github, runs, linux_only, now) + jobs_by_run, unread, unlisted = read_run_jobs(github, runs, linux_only, now) + listing = listing_state(github, unlisted) # A run whose jobs could not be read: the snapshot below is still written from every other run, # and then the sweep decides nothing (no cancel, re-run or orphan), since a partial inventory # misjudges the others too (a newer run at the same head, the pool's wait p90). @@ -3437,12 +3553,16 @@ def read_heads(branches_wanted: Sequence[str]) -> None: print(f"queue-janitor: {error}", file=sys.stderr) print(render_rate_limits(github.rate_limits), file=sys.stderr) return 1 - if unread: + # How much of the queue this sweep saw: `truncated` makes every pool reader take the queue as unknown. + snapshot.update({key: listing[key] for key in ("truncated", "listed", "total")}) + snapshot["listing"] = {key: listing[key] for key in ("jobs_unlisted", "statuses")} + if unread or listing["truncated"]: # Read from every run but these: their jobs are missing, so a label one of them is running on # can look starved and one it waits on can look recovered. No starved-label issue is opened, # updated or closed on this sweep (starved_labels.py), and readers can tell it is partial. snapshot["starved"] = [] snapshot["starved_incomplete"] = True + if unread: snapshot[UNREAD_RUNS_KEY] = sorted(unread) # Warm affinity on: which root runner kept a build of which main # commits (owned_warm_state.py). A failure leaves `warm` out, and the @@ -3462,14 +3582,22 @@ def read_heads(branches_wanted: Sequence[str]) -> None: print(f"queue-janitor: fleet gates: {gates_error}", file=sys.stderr) args.pool_load.write_text(json.dumps(snapshot, indent=2) + "\n", encoding="utf-8") - if unread: + if unread or listing["truncated"]: # Nothing is acted on, so the pending re-runs an earlier sweep left are carried forward as they were. if snapshot is not None and args.pool_load and pending is not None: snapshot[LOST_RERUNS_KEY] = [dict(entry) for entry in pending][-LOST_PENDING_MAX:] args.pool_load.write_text(json.dumps(snapshot, indent=2) + "\n", encoding="utf-8") - summary = (f"\n_The job listings of {len(unread)} run(s) failed ({', '.join(map(str, sorted(unread)))}): " - f"the pool snapshot was written from the other {len(runs)}, and this sweep decided nothing._\n" - f"\n_{render_rate_limits(github.rate_limits)}_\n") + if unread: + summary = (f"\n_The job listings of {len(unread)} run(s) failed ({', '.join(map(str, sorted(unread)))}): " + f"the pool snapshot was written from the other {len(runs)}, and this sweep decided nothing._\n") + else: + summary = "" + if listing["truncated"]: + # An annotation too: the ceilings (MAX_RUN_PAGES, MAX_JOB_LISTINGS) need raising. + print(f"::warning title=queue janitor listing truncated::{render_listing(listing)}") + summary += (f"\n_{render_listing(listing)}: the pool snapshot says `truncated`, and this sweep " + "decided nothing._\n") + summary += f"\n_{render_rate_limits(github.rate_limits)}_\n" print(summary) if args.summary: with args.summary.open("a", encoding="utf-8") as handle: @@ -3558,6 +3686,7 @@ def save_lost_reruns(entries: Sequence[Mapping[str, Any]]) -> None: summary += render_orphan_summary(orphan_decisions, dry_run=args.dry_run, now=now, min_age=orphan_age, results=orphan_results) summary += render_superseded_summary(superseded_decisions) + summary += f"\n_{render_listing(listing)}._\n" summary += f"\n_{len(runs)} in-flight runs, {len(jobs_by_run)} job listings, {len(branches)} PR branches, " \ f"{len(orphans)} orphans, {len(lost)} lost assignments, {len(unclaimed)} unclaimed legs, " \ f"{head_reads} branch head reads, {len(evidence_jobs)} evidence job listings, " \ diff --git a/scripts/ci/runner_pools.py b/scripts/ci/runner_pools.py index 884e4577d740..5600ccf3b17d 100644 --- a/scripts/ci/runner_pools.py +++ b/scripts/ci/runner_pools.py @@ -220,6 +220,8 @@ def label_evidence(health: Mapping[str, Any] | None, snapshot: Mapping[str, Any] elif generated is not None and not (dt.timedelta(minutes=-1) <= now - generated <= SNAPSHOT_MAX_AGE): notes.append(f"the janitor snapshot is from {iso(generated)}, outside the last {SNAPSHOT_MAX_AGE}; " "no snapshot evidence") + elif generated is not None and pr_runner_pool.snapshot_truncated(snapshot): + notes.append(f"{pr_runner_pool.snapshot_truncated(snapshot)}; no snapshot evidence") elif generated is not None: age = max(0.0, (now - generated).total_seconds() / 60) for label, load in (snapshot.get("pools") or {}).items(): From fcdbc2d61ed5dfec1ab2863e9e72e65f44ff780b Mon Sep 17 00:00:00 2001 From: Leo Li Date: Fri, 30 Oct 2026 15:50:57 +0000 Subject: [PATCH 3/3] ci(janitor): requests in flight when the App token is refused retry on the workflow token The job listings now run on several threads. A request sent with the App token before another thread switched to the workflow token found no fallback left and failed, which marked its run unread and made the whole sweep decide nothing. The switch is made under a lock, and a request refused with a token that is no longer current is made again with the current one. Co-Authored-By: Claude Opus 5.5 Agent-Session: https://cmux.com/deeplink/agent?id=aid-0b4091cd6a9b72e19edf --- scripts/ci/queue_janitor.py | 26 ++++++++++++++++--------- tests/test_ci_queue_janitor.py | 35 ++++++++++++++++++++++++++++++++++ 2 files changed, 52 insertions(+), 9 deletions(-) diff --git a/scripts/ci/queue_janitor.py b/scripts/ci/queue_janitor.py index 12f264e61bbd..5e71468cecc1 100644 --- a/scripts/ci/queue_janitor.py +++ b/scripts/ci/queue_janitor.py @@ -3042,15 +3042,23 @@ def _request(self, method: str, path: str, body: Any | None = None, *, app_only: if error.code in (403, 429) and rate_limited(error): raise RateLimited(f"{method} {path.split('?')[0]} failed ({error.code}: rate limit; " "the sweep stops rather than spend the workflow token)") from error - if error.code in (401, 403) and self.fallback_token and not app_only: - print(f"queue-janitor: App token refused ({error.code}) on {path.split('?')[0]}; " - "using the workflow token for the rest of the sweep", file=sys.stderr) - self.headers["Authorization"] = f"Bearer {self.fallback_token}" - # The snapshot download (download()) uses the same token. - self._token = self.fallback_token - self.fallback_token = "" - self.app_token = False - return self.request(method, path, body) + if error.code in (401, 403) and not app_only: + # The job listings run on several threads (read_run_jobs()): one switches the token, and a + # request another sent with the refused token before the switch is made again with the new + # one, instead of failing for want of a fallback the first thread already used. + with self._count_lock: + switched = headers.get("Authorization") != self.headers.get("Authorization") + if not switched and self.fallback_token: + print(f"queue-janitor: App token refused ({error.code}) on {path.split('?')[0]}; " + "using the workflow token for the rest of the sweep", file=sys.stderr) + self.headers["Authorization"] = f"Bearer {self.fallback_token}" + # The snapshot download (download()) uses the same token. + self._token = self.fallback_token + self.fallback_token = "" + self.app_token = False + switched = True + if switched: + return self.request(method, path, body) if error.code in RETRY_STATUSES: raise _Transient(str(error.code)) from error raise RuntimeError(f"{method} {path.split('?')[0]} failed ({error.code})") from error diff --git a/tests/test_ci_queue_janitor.py b/tests/test_ci_queue_janitor.py index 1eee684d5bea..f62b6f81b149 100644 --- a/tests/test_ci_queue_janitor.py +++ b/tests/test_ci_queue_janitor.py @@ -2029,6 +2029,41 @@ def urlopen(request, timeout=30): with mock.patch.object(janitor.urllib.request, "urlopen", urlopen), self.assertRaises(RuntimeError): github.request("GET", "/repos/manaflow-ai/cmux/actions/runs") + def test_requests_in_flight_when_the_app_token_is_refused_all_move_to_the_workflow_token(self): + # The job listings run on several threads: every request sent with the App token before one of them + # switched to the workflow token is made again with it, not failed for want of a fallback. + import threading + github = janitor.GitHub("app", "manaflow-ai/cmux", fallback_token="workflow") + both_sent = threading.Barrier(4, timeout=5) + sent = [] + + class Response(io.BytesIO): + headers = email.message.Message() + + def urlopen(request, timeout=30): + token = request.get_header("Authorization") + sent.append(token) + if token == "Bearer app": + both_sent.wait() # all four are in flight with the App token before any is refused + raise urllib.error.HTTPError(request.full_url, 401, "bad credentials", email.message.Message(), None) + return Response(b'{"jobs": []}') + + runs = [make_run(status="queued", age=age, branch=f"b{age}") for age in range(1, 5)] + with mock.patch.object(janitor.urllib.request, "urlopen", urlopen), mock.patch("builtins.print"), \ + mock.patch.object(janitor.time, "sleep", lambda _: None): + jobs, unread, _ = janitor.read_run_jobs(github, runs, frozenset(), NOW, workers=4) + self.assertEqual(unread, set()) + self.assertEqual(set(jobs), {run["id"] for run in runs}) + self.assertEqual(sorted(sent), ["Bearer app"] * 4 + ["Bearer workflow"] * 4) + # A refusal of the workflow token itself is still an error, not a loop. + refused = janitor.GitHub("app", "manaflow-ai/cmux", fallback_token="workflow") + + def always(request, timeout=30): + raise urllib.error.HTTPError(request.full_url, 401, "bad credentials", email.message.Message(), None) + with mock.patch.object(janitor.urllib.request, "urlopen", always), mock.patch("builtins.print"), \ + self.assertRaises(RuntimeError): + refused.request("GET", "/repos/manaflow-ai/cmux/actions/runs/1/jobs") + def test_an_app_out_of_budget_stops_the_sweep_instead_of_spending_the_workflow_token(self): import email.message import urllib.error