ci: read the gui backlog eight runs at a time in late placement

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
@@ -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
 
@@ -108,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,
@@ -123,13 +126,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
 
 

tests/test_ci_late_placement.py

@@ -230,11 +230,55 @@ 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
+        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):
+                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):
+                together.wait()
+                with self.lock:
+                    self.jobs_read.append(int(path.split("/")[3]))
+                return {"jobs": [{"status": "queued", "labels": [GUI]}]}
+        api = API()
+        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, and nothing after `enough`.
+        self.assertEqual(sorted(api.jobs_read), list(range(1, late.BACKLOG_READERS + 1)))
+
+    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):