feat(ci): the workflow runner drains through a bounded pool, not serially #5510
2 changed files+115−6
Addedsrc/__tests__/workflow-runner-concurrency.test.ts+65−0View fileUnifiedSplit
@@ -0,0 +1,65 @@
1/**
2 * Runner concurrency (2026-08-22, scorecard move #2).
3 *
4 * The queue was strictly serial: one slow run stalled every other repo's CI
5 * behind it. startWorker now drains through a bounded pool. The behavioral
6 * surface lives inside the tick closure, so — following the house style of
7 * workflow-run-abandonment.test.ts — the pool's load-bearing properties are
8 * pinned against the source, and the pure concurrency resolver is tested
9 * directly.
10 */
11
12import { describe, expect, test } from "bun:test";
13import { resolveWorkflowConcurrency } from "../lib/workflow-runner";
14
15describe("resolveWorkflowConcurrency", () => {
16 test("defaults to 3", () => {
17 expect(resolveWorkflowConcurrency({})).toBe(3);
18 });
19
20 test("honors WORKFLOW_CONCURRENCY", () => {
21 expect(resolveWorkflowConcurrency({ WORKFLOW_CONCURRENCY: "5" })).toBe(5);
22 });
23
24 test("clamps garbage and extremes — a typo can't fork-bomb the container", () => {
25 expect(resolveWorkflowConcurrency({ WORKFLOW_CONCURRENCY: "0" })).toBe(1);
26 expect(resolveWorkflowConcurrency({ WORKFLOW_CONCURRENCY: "-4" })).toBe(1);
27 expect(resolveWorkflowConcurrency({ WORKFLOW_CONCURRENCY: "banana" })).toBe(1);
28 expect(resolveWorkflowConcurrency({ WORKFLOW_CONCURRENCY: "500" })).toBe(16);
29 expect(resolveWorkflowConcurrency({ WORKFLOW_CONCURRENCY: "2.9" })).toBe(2);
30 });
31});
32
33describe("the tick drains through a bounded pool", () => {
34 async function tickSource(): Promise<string> {
35 const src = await Bun.file("src/lib/workflow-runner.ts").text();
36 const code = src
37 .split("\n")
38 .filter((l) => !l.trimStart().startsWith("*") && !l.trimStart().startsWith("//"))
39 .join("\n");
40 return code.slice(code.indexOf("const tick = async"));
41 }
42
43 test("slots are capped by the resolved concurrency", async () => {
44 const tick = await tickSource();
45 expect(tick).toContain("inFlight.size < concurrency");
46 });
47
48 test("drainOneRun stays the claim-execute unit (atomic claim intact)", async () => {
49 const tick = await tickSource();
50 expect(tick).toContain("drainOneRun()");
51 });
52
53 test("an empty queue stops refilling instead of spinning the DB", async () => {
54 const tick = await tickSource();
55 const latch = tick.indexOf("sawEmptyQueue = true");
56 const gate = tick.indexOf("!sawEmptyQueue && inFlight.size");
57 expect(latch).toBeGreaterThan(-1);
58 expect(gate).toBeGreaterThan(-1);
59 });
60
61 test("the tick waits for a slot with race, not a fixed sleep", async () => {
62 const tick = await tickSource();
63 expect(tick).toContain("Promise.race(inFlight)");
64 });
65});
Modifiedsrc/lib/workflow-runner.ts+50−6View fileUnifiedSplit
@@ -1196,8 +1196,26 @@ export async function enqueueRun(opts: {
11961196// Public: startWorker — background poll loop.
11971197// ---------------------------------------------------------------------------
11981198
1199export function startWorker(opts?: { intervalMs?: number }): () => void {
1199/**
1200 * How many runs may execute at once. One box, and the runner lives inside
1201 * the app container — so the default is deliberately small; this is
1202 * "stop being a strictly serial queue", not "be a build farm". Clamped so a
1203 * typo'd env value can't fork-bomb the container.
1204 */
1205export function resolveWorkflowConcurrency(
1206 env: Record<string, string | undefined> = process.env
1207): number {
1208 const n = Number(env.WORKFLOW_CONCURRENCY || 3);
1209 if (!Number.isFinite(n) || n < 1) return 1;
1210 return Math.min(Math.floor(n), 16);
1211}
1212
1213export function startWorker(opts?: {
1214 intervalMs?: number;
1215 concurrency?: number;
1216}): () => void {
12001217 const intervalMs = opts?.intervalMs ?? DEFAULT_POLL_INTERVAL_MS;
1218 const concurrency = opts?.concurrency ?? resolveWorkflowConcurrency();
12011219 const bootTime = new Date();
12021220 let stopped = false;
12031221 let active = false;
@@ -1239,11 +1257,37 @@ export function startWorker(opts?: { intervalMs?: number }): () => void {
12391257 }
12401258 await reapStuckRuns();
12411259
1242 // Drain as many runs as we can in one tick (serial). If there's
1243 // nothing queued we exit quickly and wait for the next interval.
1244 let picked = true;
1245 while (picked && !stopped) {
1246 picked = await drainOneRun();
1260 // Drain the queue through a bounded pool (WORKFLOW_CONCURRENCY,
1261 // default 3; was strictly serial until 2026-08-22 — one slow run
1262 // stalled every other repo's CI behind it). Each drainOneRun() claims
1263 // atomically (queued→running conditional UPDATE), so concurrent slots
1264 // can never double-execute a run — the claim was always
1265 // multi-worker-safe; there was just never a second caller.
1266 //
1267 // `sawEmptyQueue` stops slot-refilling once a claim comes back empty,
1268 // and is reset only when a slot that DID work completes — without that
1269 // gate, an empty queue would spin claim-roundtrips against the DB for
1270 // as long as any long run stayed in flight.
1271 const inFlight = new Set<Promise<boolean>>();
1272 let sawEmptyQueue = false;
1273 while (!stopped) {
1274 while (!sawEmptyQueue && inFlight.size < concurrency) {
1275 // Promise.resolve().then defers the body one microtask, so `slot`
1276 // is assigned before the closure (and its finally) can run.
1277 const slot: Promise<boolean> = Promise.resolve().then(async () => {
1278 try {
1279 const worked = await drainOneRun();
1280 if (!worked) sawEmptyQueue = true;
1281 return worked;
1282 } finally {
1283 inFlight.delete(slot);
1284 }
1285 });
1286 inFlight.add(slot);
1287 }
1288 if (inFlight.size === 0) break;
1289 const worked = await Promise.race(inFlight);
1290 if (worked) sawEmptyQueue = false;
12471291 }
12481292 } catch (err) {
12491293 console.error("[workflow-runner] worker tick:", err);
12501294
c comment · e edit title · m merge · a approve · r request changes · ? shortcuts