From 97347d380c45dbbdea3d34121fdd3e1c4dfd4a45 Mon Sep 17 00:00:00 2001 From: Leo Li Date: Mon, 28 Sep 2026 03:28:19 -0400 Subject: [PATCH 1/2] ci: read the gui backlog eight runs at a time in late placement late_placement.py counted queued gui jobs by reading up to 30 in-flight runs' job lists one by one. Under load that outran the step's one-minute timeout in 10 of the last 21 PR runs; the step then wrote no placement, so every shard stayed queued on the owned gui runners instead of moving its overflow to Blacksmith. Reads now go eight at a time, oldest batch first, stopping once the count reaches what decide() compares against. Co-Authored-By: Claude Opus 5.5 (1M context) --- scripts/ci/late_placement.py | 20 +++++++++++++------ tests/test_ci_late_placement.py | 35 ++++++++++++++++++++++++++++++--- 2 files changed, 46 insertions(+), 9 deletions(-) diff --git a/scripts/ci/late_placement.py b/scripts/ci/late_placement.py index eba926cd7816..4f3383d8e747 100644 --- a/scripts/ci/late_placement.py +++ b/scripts/ci/late_placement.py @@ -44,6 +44,7 @@ """ from __future__ import annotations +import concurrent.futures import datetime as dt import importlib.util import json @@ -71,6 +72,7 @@ def _picker(): # jobs queue only once its admission finished (p50 about 9 minutes), so a run younger than BACKLOG_MIN_AGE has none, # and one older than the window has finished its shards. BACKLOG_LOOKUPS = 30 +BACKLOG_READERS = 8 BACKLOG_WINDOW_MINUTES = 120 BACKLOG_MIN_AGE_MINUTES = 4 @@ -123,13 +125,19 @@ def gui_backlog(github: Any, label: str, *, exclude_run_id: int | None, enough: created = pool.parse_time(str(run.get("created_at") or "")) if run.get("id") != exclude_run_id and created is not None and created <= newest: runs[run.get("id")] = run - queued = 0 - for run in sorted(runs.values(), key=lambda run: str(run.get("created_at")))[:BACKLOG_LOOKUPS]: - if queued >= enough: - break + def queued_in(run: Mapping[str, Any]) -> int: jobs = github.get(f"/actions/runs/{run['id']}/jobs?filter=latest&per_page={pool.PAGE_SIZE}").get("jobs") or [] - queued += sum(1 for job in jobs if isinstance(job, Mapping) and job.get("status") == "queued" - and label in (job.get("labels") or [])) + return sum(1 for job in jobs if isinstance(job, Mapping) and job.get("status") == "queued" + and label in (job.get("labels") or [])) + + # Read BACKLOG_READERS runs at a time: one by one, 30 job lists under load outran the step's minute. + ordered = sorted(runs.values(), key=lambda run: str(run.get("created_at")))[:BACKLOG_LOOKUPS] + queued = 0 + with concurrent.futures.ThreadPoolExecutor(max_workers=BACKLOG_READERS) as readers: + for start in range(0, len(ordered), BACKLOG_READERS): + if queued >= enough: + break + queued += sum(readers.map(queued_in, ordered[start:start + BACKLOG_READERS])) return queued diff --git a/tests/test_ci_late_placement.py b/tests/test_ci_late_placement.py index fd2da95e0033..b1c90e18f800 100644 --- a/tests/test_ci_late_placement.py +++ b/tests/test_ci_late_placement.py @@ -230,12 +230,41 @@ def get(self, path): return {"jobs": [{"status": "queued", "labels": [GUI]}, {"status": "queued", "labels": [ROOT_STD]}, {"status": "in_progress", "labels": [GUI]}, {"status": "queued", "labels": [GUI]}]} api = API() - self.assertEqual(late.gui_backlog(api, GUI, exclude_run_id=7, enough=5, now=now), 6) - # Oldest first; run 4 is too young to have gui jobs and 7 is this run. - self.assertEqual(api.jobs_read, [8, 6, 5]) + # One batch of readers takes every eligible run, so the count may pass `enough` (decide() only + # compares against it). Run 4 is too young to have gui jobs and 7 is this run. + self.assertEqual(late.gui_backlog(api, GUI, exclude_run_id=7, enough=5, now=now), 8) + self.assertEqual(sorted(api.jobs_read), [5, 6, 8, 9]) self.assertEqual(api.statuses, [("ci.yml", "queued"), ("ci.yml", "in_progress")]) + def test_backlog_reads_runs_concurrently_oldest_batch_first_and_stops_when_enough(self): + import datetime as dt + import threading + import time + now = dt.datetime(2026, 9, 28, 1, 0, tzinfo=dt.timezone.utc) + runs = [{"id": i, "created_at": (now - dt.timedelta(minutes=100 - i)).strftime("%Y-%m-%dT%H:%M:%SZ")} + for i in range(1, 21)] + + class API: + def __init__(self): + self.jobs_read, self.lock = [], threading.Lock() + + def runs_since(self, workflow, since, **filters): + return runs if filters["status"] == "in_progress" else [] + + def get(self, path): + time.sleep(0.2) + with self.lock: + self.jobs_read.append(int(path.split("/")[3])) + return {"jobs": [{"status": "queued", "labels": [GUI]}]} + api = API() + started = time.monotonic() + self.assertEqual(late.gui_backlog(api, GUI, exclude_run_id=None, enough=late.BACKLOG_READERS, now=now), + late.BACKLOG_READERS) + # One batch: the oldest BACKLOG_READERS runs, read at once, and nothing after `enough`. + self.assertEqual(sorted(api.jobs_read), list(range(1, late.BACKLOG_READERS + 1))) + self.assertLess(time.monotonic() - started, 0.2 * late.BACKLOG_READERS / 2) + class Output(unittest.TestCase): def test_main_writes_an_empty_object_without_a_token(self): import tempfile From ca95bb7ec20f4cf78beeee82508f5bae6491339e Mon Sep 17 00:00:00 2001 From: Leo Li Date: Mon, 28 Sep 2026 03:35:47 -0400 Subject: [PATCH 2/2] ci: prove the backlog reads overlap with a barrier, not a clock; test a failed read Co-Authored-By: Claude Opus 5.5 (1M context) --- scripts/ci/late_placement.py | 5 +++-- tests/test_ci_late_placement.py | 27 +++++++++++++++++++++------ 2 files changed, 24 insertions(+), 8 deletions(-) diff --git a/scripts/ci/late_placement.py b/scripts/ci/late_placement.py index 4f3383d8e747..c4aba87b4563 100644 --- a/scripts/ci/late_placement.py +++ b/scripts/ci/late_placement.py @@ -31,7 +31,7 @@ Blacksmith (about 350 s against 240 to 400 s), so waiting for a mini buys nothing. When the picker owned some of this run's gui-token jobs and the gui runners idle now cannot take them all, this counts the gui-label jobs already -queued (gui_backlog(): the jobs of in-flight CI runs, newest runs first, +queued (gui_backlog(): the jobs of in-flight CI runs, oldest runs first, stopping once the answer cannot change). The run keeps on the gui label only the jobs that start within GUI_QUEUE_ROUNDS gui job lengths (the idle runners plus that many rounds of the online ones, less the backlog), and gives the @@ -110,7 +110,8 @@ def place(jobs: Sequence[str], *, owned_jobs: str, idle: int, root: str, gui: bo def gui_backlog(github: Any, label: str, *, exclude_run_id: int | None, enough: int, now: dt.datetime) -> int: - """Jobs queued on `label` in the CI runs still in flight, stopping at `enough` (one request per run). + """Jobs queued on `label` in the CI runs still in flight (one request per run), read BACKLOG_READERS runs + at a time and stopping after the batch that reaches `enough`, so the count may pass it. GitHub lists a run as `queued` while any of its jobs is, even with others running, so both `queued` and `in_progress` runs are read. Oldest first, diff --git a/tests/test_ci_late_placement.py b/tests/test_ci_late_placement.py index b1c90e18f800..97fded7a5c9b 100644 --- a/tests/test_ci_late_placement.py +++ b/tests/test_ci_late_placement.py @@ -236,14 +236,14 @@ def get(self, path): self.assertEqual(sorted(api.jobs_read), [5, 6, 8, 9]) self.assertEqual(api.statuses, [("ci.yml", "queued"), ("ci.yml", "in_progress")]) - def test_backlog_reads_runs_concurrently_oldest_batch_first_and_stops_when_enough(self): import datetime as dt import threading - import time now = dt.datetime(2026, 9, 28, 1, 0, tzinfo=dt.timezone.utc) runs = [{"id": i, "created_at": (now - dt.timedelta(minutes=100 - i)).strftime("%Y-%m-%dT%H:%M:%SZ")} for i in range(1, 21)] + # Every read of a batch must be in flight at once, or the barrier breaks and the read raises. + together = threading.Barrier(late.BACKLOG_READERS, timeout=30) class API: def __init__(self): @@ -253,17 +253,32 @@ def runs_since(self, workflow, since, **filters): return runs if filters["status"] == "in_progress" else [] def get(self, path): - time.sleep(0.2) + together.wait() with self.lock: self.jobs_read.append(int(path.split("/")[3])) return {"jobs": [{"status": "queued", "labels": [GUI]}]} api = API() - started = time.monotonic() self.assertEqual(late.gui_backlog(api, GUI, exclude_run_id=None, enough=late.BACKLOG_READERS, now=now), late.BACKLOG_READERS) - # One batch: the oldest BACKLOG_READERS runs, read at once, and nothing after `enough`. + # One batch: the oldest BACKLOG_READERS runs, and nothing after `enough`. self.assertEqual(sorted(api.jobs_read), list(range(1, late.BACKLOG_READERS + 1))) - self.assertLess(time.monotonic() - started, 0.2 * late.BACKLOG_READERS / 2) + + def test_a_failed_backlog_read_raises_so_nothing_moves(self): + import datetime as dt + now = dt.datetime(2026, 9, 28, 1, 0, tzinfo=dt.timezone.utc) + + class API: + def runs_since(self, workflow, since, **filters): + return [{"id": i, "created_at": (now - dt.timedelta(minutes=30 + i)).strftime("%Y-%m-%dT%H:%M:%SZ")} + for i in range(1, 4)] if filters["status"] == "in_progress" else [] + + def get(self, path): + if path.split("/")[3] == "2": + raise RuntimeError("HTTP 502") + return {"jobs": []} + with self.assertRaises(RuntimeError): + late.gui_backlog(API(), GUI, exclude_run_id=None, enough=5, now=now) + class Output(unittest.TestCase): def test_main_writes_an_empty_object_without_a_token(self):