trevato/forge

Public: anyone can read this; the crew can change it.

More
Forks1g fMapg aRuns15g rEventsFlows

The merge queue gates in parallel: bisect a red batch two halves at once, and speculatively gate the next batch on top of the current one #4058

Merged masoncrew opened this in answer to trevatoowner

Check · Review · Land · Live done — Landed and live in production.

ConversationFiles
Merged

mason/forge:mason/4053-the-merge-queue-gates-in-parallel-bisect → main 6 commits fork

  • Merged. mason/4053-the-merge-queue-gates-in-parallel-bisect was merged into main as bc6bb89.

Review

Sign in to review this change.
10 files changed +881 −156
README.md +17 −5
@@ -1221,8 +1221,17 @@ The log says `[queue] gating N pulls since HH:MM` as a gate starts and
12211221 The full gate runs once on that candidate. Green lands every pull in order,
12221222 and each landing is refused unless main is still where the candidate was
12231223 built and the squash writes the tree that was gated, so main ends at the
1224 gated tree. Red bisects: the front half is gated by itself, a green half
1225 lands, and the red pull is handed back with the failing tests named. A red
1224 gated tree. Red is searched: prefixes of the candidate are gated on main
1225 — the front half, plus one more cut per free check slot, all at once — the
1226 longest green prefix lands, the search goes on in the steps up to the first
1227 red prefix, and the red pull is handed back with the failing tests named.
1228 While a candidate gates, the next one gates on top of it (`queue.depth`
1229 under Settings → Capacity, `FORGE_QUEUE_DEPTH`, default 2): when the first
1230 lands whole the next lands right after it, on the tree it was gated on; when
1231 it does not, the next is formed again on main. A gate beyond a line's first
1232 starts only on a free check slot and holds it, and two slots are always kept
1233 for pull checks (`PULL_RESERVE`); with no slot spare the queue gates one
1234 candidate at a time and halves a red one, as it always did. A red
12261235 waits for a push or a new main, except once: after 30 minutes with neither
12271236 side moved it is gated again, and lands if green; a red that reproduces
12281237 stays red and is not told twice. A gate cut off by the forge stopping (a
@@ -1231,8 +1240,9 @@ sweep gates it again. So is a gate that has not answered in 15 minutes
12311240 (`GATE_DEADLINE_MS`): it is abandoned and its check aborted, the log says
12321241 `[queue] candidate abandoned`, and the next sweep forms the line again. Pulls
12331242 that turn green while a candidate is gating wait for the next one. Each
1234 queue's numbers — how many pulls are queued, how many are gating, and how
1235 many landed in the last hour — show in three places: the flow panel on `/boards`, the
1243 queue's numbers — how many pulls are queued, how many are gating, how many
1244 gates are in flight, how many landed in the last hour, and their median
1245 minutes from approval to landing — show in three places: the flow panel on `/boards`, the
12361246 terminal's Flow command, and the MCP `flow` tool's `queue` field.
12371247 When a repository requires checks, an approved pull that is green only
12381248 against an older main joins the queue as it is: the candidate's full gate
@@ -1241,7 +1251,9 @@ its own would give, and a red candidate bisects as above. So a landing that
12411251 moves main does not send every other approved pull off for a check of its
12421252 own first; the batch lands in one gate. A pull's own check runs again only
12431253 when its head changes. A person's merge, which no candidate gated, still
1244 refuses a stale green. The autonomy tick line counts the approved pulls the
1254 refuses a stale green. Each sweep reads every open pull, oldest first, not
1255 only the newest page, so an older approved pull is batched with the rest
1256 rather than joining alone on its own event. The autonomy tick line counts the approved pulls the
12451257 last sweep saw and why each waits (`approved="3 approved: queued=2 review=1"`,
12461258 from `src/crew/waits.ts`), and what the queue is doing now
12471259 (`queue="gating 2 pulls since 18:54"`, `queue="1 waiting"` or `queue=idle`,
src/checks/queue.ts +385 −132
@@ -1,23 +1,31 @@
11 /**
22 * The merge queue (#2036): per repository, one line of pulls that are green
3 * and approved. The line is built into ONE candidate — the base's tip, then
3 * and approved. The line is built into candidates — the base's tip, then
44 * each queued pull's squash on top of the one before, in queue order, for
5 * every pull that merges cleanly — and that candidate is gated once. Green
5 * every pull that merges cleanly — and each candidate is gated once. Green
66 * lands every step in order, each landing guarded so the squash it writes
77 * is the step's tree on the base the step was built on: main ends at
8 * exactly the tree that was gated. Red bisects: the first half is gated on
9 * its own; green lands it and the culprit is in the rest, red keeps the
10 * search in the first half while the rest waits for the next candidate. A
11 * single red step is the culprit and is handed back. A gate cut off by the
12 * forge stopping is no verdict: nothing is handed back, and the forge that
13 * carries on admits the line again (#3155). A gate that has not answered
14 * by `GATE_DEADLINE_MS` is abandoned the same way: cut off, logged, and the
15 * line re-formed by the next sweep, so one gate that never returns cannot
16 * hold the queue for good (#3430).
8 * exactly the tree that was gated.
9 *
10 * The line gates in parallel where check slots are free (#4053). While a
11 * candidate gates on main, the next gates on top of it (`depth` deep): if
12 * the first lands whole, the next lands right after on the tree it was
13 * gated on; if not, the next is discarded and formed again on main. A red
14 * candidate is searched by gating several of its prefixes at once, one per
15 * free slot: the longest green prefix lands, the culprit is in the steps up
16 * to the first red prefix, and the steps after it wait for the next
17 * candidate. With no slot free that is one prefix, the front half: plain
18 * bisection. A single red step is the culprit and is handed back.
19 *
20 * A gate cut off by the forge stopping is no verdict: nothing is handed
21 * back, and the forge that carries on admits the line again (#3155). A gate
22 * that has not answered by `GATE_DEADLINE_MS` is abandoned the same way:
23 * cut off, logged, and the line re-formed by the next sweep, so one gate
24 * that never returns cannot hold the queue for good (#3430).
1725 *
1826 * What is queued is not stored: it is the pulls `landIfGreen` admitted, and
1927 * every sweep admits them again. The queue only orders them and remembers
20 * how many there are, for the flow panel on /boards.
28 * what is at the gate, for the flow panel on /boards.
2129 */
2230
2331 import type { Pull } from "../core/pulls.ts";
@@ -59,12 +67,15 @@ export interface Step {
5967 readonly entry: Queued;
6068 readonly baseOid: string;
6169 readonly treeOid: string;
70 /** The chain's commit once this step is on it: the tip of its prefix. */
71 readonly commitOid: string;
6272 }
6373
6474 /** What is gated: a chain of steps on the base, ending at `tipOid`. */
6575 export interface Candidate {
6676 readonly dir: string;
6777 readonly baseOid: string;
78 readonly baseTreeOid: string;
6879 readonly steps: readonly Step[];
6980 readonly tipOid: string;
7081 readonly treeOid: string;
@@ -92,6 +103,20 @@ let deadlineMs = GATE_DEADLINE_MS;
92103 */
93104 export const MAX_CANDIDATE = 8;
94105
106 /** Candidates gating at once by default: one on main, the next on top of it. */
107 export const DEFAULT_DEPTH = 2;
108
109 /** How the line may use the forge's check slots (#4053). */
110 export interface QueueOptions {
111 /** How many candidates may gate at once, each on the one before. */
112 readonly depth?: number;
113 /**
114 * Check slots free now for a gate beyond a line's first, the pull
115 * checks' reserve already kept back; none when not said.
116 */
117 readonly room?: () => number;
118 }
119
95120 /** Tests shorten the deadline; the answer puts it back. */
96121 export function gateDeadlineForTest(ms: number): () => void {
97122 deadlineMs = ms;
@@ -100,32 +125,63 @@ export function gateDeadlineForTest(ms: number): () => void {
100125 };
101126 }
102127
103 /** The gate as a round asks it: the deadline is the line's own. */
104 type Gated = (candidate: Candidate) => Promise<LandVerdict>;
128 /** The gate as the line asks it: the deadline is the line's own. */
129 type Gated = (candidate: Candidate, signal?: AbortSignal) => Promise<LandVerdict>;
130
131 /** Check slots free now for one more gate beyond a line's first. */
132 type Room = () => number;
105133
106134 interface Waiting {
107135 readonly entry: Queued;
108136 readonly settle: (landed: boolean) => void;
109137 }
110138
139 /** One gate in flight: the pulls its candidate carries, and since when. */
140 interface InFlight {
141 readonly pulls: readonly string[];
142 readonly since: number;
143 }
144
111145 interface Line {
112146 readonly repoId: string;
113147 waiting: Waiting[];
114 /** Pull ids in the queue or its candidate: a pull is in the line once. */
148 /** Pull ids in the queue or a candidate: a pull is in the line once. */
115149 readonly holds: Set<string>;
116 /** How many pulls the candidate at the gate now carries. */
117 gating: number;
118 /** When the candidate at the gate started, epoch ms; 0 when none is. */
119 since: number;
150 /** The gates this line has in flight now. */
151 readonly flights: Set<InFlight>;
120152 /** The gate runs this line has made, for a test and the log. */
121153 gates: number;
122154 draining: boolean;
155 /** Wakes a drain waiting on a gate when pulls join the line. */
156 wake: (() => void) | undefined;
157 }
158
159 /** A candidate at the gate in the pipeline, and the pulls it took from the line. */
160 interface Flight {
161 readonly steps: readonly Waiting[];
162 readonly candidate: Candidate;
163 /** Built on the flight before it, not on the base's tip. */
164 readonly speculative: boolean;
165 readonly branch: string;
166 readonly verdict: Promise<LandVerdict>;
167 readonly abort: AbortController;
123168 }
124169
125170 const log = createLog("queue");
126171
127172 const lines = new Map<string, Line>();
128173
174 /** Gates in flight beyond each line's first, every line together. */
175 let extra = 0;
176
177 /**
178 * Gates the queue has in flight beyond each line's first: each holds a
179 * check slot a pull check does not take (`checks/service.ts`).
180 */
181 export function extraGates(): number {
182 return extra;
183 }
184
129185 const IDENTITY = {
130186 GIT_AUTHOR_NAME: "forge",
131187 GIT_AUTHOR_EMAIL: "forge@localhost",
@@ -133,10 +189,12 @@ const IDENTITY = {
133189 GIT_COMMITTER_EMAIL: "forge@localhost",
134190 };
135191
192 const UNDECIDED: LandVerdict = { lands: false, interrupted: true };
193
136194 function lineOf(repoId: string): Line {
137195 let line = lines.get(repoId);
138196 if (line === undefined) {
139 line = { repoId, waiting: [], holds: new Set(), gating: 0, since: 0, gates: 0, draining: false };
197 line = { repoId, waiting: [], holds: new Set(), flights: new Set(), gates: 0, draining: false, wake: undefined };
140198 lines.set(repoId, line);
141199 }
142200 return line;
@@ -151,6 +209,7 @@ export function enqueue(
151209 repoId: string,
152210 entries: readonly Queued[],
153211 gate: GateCandidate,
212 options: QueueOptions = {},
154213 ): Promise<boolean[]> {
155214 const line = lineOf(repoId);
156215 const answers = entries.map((entry) => {
@@ -166,7 +225,11 @@ export function enqueue(
166225 }),
167226 );
168227 });
169 if (!line.draining) void drain(line, gate);
228 if (!line.draining) void drain(line, gate, options);
229 else {
230 line.wake?.();
231 line.wake = undefined;
232 }
170233 return Promise.all(answers);
171234 }
172235
@@ -174,32 +237,43 @@ export function enqueue(
174237 export interface QueueStatus {
175238 readonly repoId: string;
176239 readonly queued: number;
240 /** Pulls in a candidate at the gate, each counted once. */
177241 readonly gating: number;
178 /** When the candidate at the gate started, epoch ms; 0 when none is. */
242 /** Gates in flight. */
243 readonly gates: number;
244 /** When the oldest gate in flight started, epoch ms; 0 when none is. */
179245 readonly since: number;
180246 }
181247
248 function statusOf(repoId: string, line: Line): QueueStatus {
249 const pulls = new Set<string>();
250 let since = 0;
251 for (const f of line.flights) {
252 for (const id of f.pulls) pulls.add(id);
253 if (since === 0 || f.since < since) since = f.since;
254 }
255 return { repoId, queued: line.waiting.length, gating: pulls.size, gates: line.flights.size, since };
256 }
257
182258 export function queueStatus(): QueueStatus[] {
183259 return [...lines.entries()]
184 .map(([repoId, line]) => ({
185 repoId,
186 queued: line.waiting.length,
187 gating: line.gating,
188 since: line.since,
189 }))
260 .map(([repoId, line]) => statusOf(repoId, line))
190261 .filter((s) => s.queued > 0 || s.gating > 0);
191262 }
192263
193264 /**
194265 * The repository's line in a few words, for the tick line (#3430):
195 * `gating 4 pulls since 18:54` (UTC), `2 waiting`, both, or `idle`.
266 * `gating 4 pulls since 18:54` (UTC), `in 2 gates` when more than one is
267 * in flight, `2 waiting`, or `idle`.
196268 */
197269 export function queueWords(repoId: string): string {
198270 const line = lines.get(repoId);
199271 if (line === undefined) return "idle";
200 const waiting = line.waiting.length === 0 ? "" : `${line.waiting.length} waiting`;
201 if (line.gating === 0) return waiting === "" ? "idle" : waiting;
202 return `gating ${pullsWord(line.gating)} since ${hhmm(line.since)}${waiting === "" ? "" : `, ${waiting}`}`;
272 const s = statusOf(repoId, line);
273 const waiting = s.queued === 0 ? "" : `${s.queued} waiting`;
274 if (s.gating === 0) return waiting === "" ? "idle" : waiting;
275 const gates = s.gates > 1 ? ` in ${s.gates} gates` : "";
276 return `gating ${pullsWord(s.gating)} since ${hhmm(s.since)}${gates}${waiting === "" ? "" : `, ${waiting}`}`;
203277 }
204278
205279 /** `1 pull`, `8 pulls`. */
@@ -224,68 +298,106 @@ export function gateRuns(repoId: string): number {
224298 }
225299
226300 /**
227 * One candidate at a time, until the line is empty. A batch is the front
228 * of the line that shares one place and one base branch; a sealed pull
229 * whose objects live elsewhere is a batch of its own. A batch carries at
230 * most `MAX_CANDIDATE` pulls.
301 * The line's gate: counted, logged as it starts and ends (a gate takes
302 * minutes, #3430), and abandoned at the deadline or when `signal` aborts —
303 * a speculative candidate discarded. Abandoned is no verdict.
231304 */
232 async function drain(line: Line, gate: GateCandidate): Promise<void> {
233 line.draining = true;
234 try {
235 while (line.waiting.length > 0) {
236 const first = line.waiting[0]!.entry;
237 const batch = line.waiting
238 .filter((w) => w.entry.dir === first.dir && w.entry.pull.baseBranch === first.pull.baseBranch)
239 .slice(0, MAX_CANDIDATE);
240 line.waiting = line.waiting.filter((w) => !batch.includes(w));
241 const counted: Gated = async (candidate) => {
242 line.gates++;
243 line.gating = candidate.steps.length;
244 line.since = Date.now();
245 const pulls = candidate.steps.map((s) => `#${s.entry.pull.number}`).join(",");
246 // Said as it starts and as it ends: a gate takes minutes (#3430).
247 log.info(`gating ${pullsWord(line.gating)} since ${hhmm(line.since)}`, {
305 function counted(line: Line, gate: GateCandidate): Gated {
306 return async (candidate, signal) => {
307 const beyond = line.flights.size > 0;
308 const flight: InFlight = { pulls: candidate.steps.map((s) => s.entry.pull.id), since: Date.now() };
309 line.flights.add(flight);
310 line.gates++;
311 if (beyond) extra++;
312 const n = candidate.steps.length;
313 const pulls = candidate.steps.map((s) => `#${s.entry.pull.number}`).join(",");
314 log.info(`gating ${pullsWord(n)} since ${hhmm(flight.since)}`, {
315 repo: line.repoId,
316 pulls,
317 waiting: line.waiting.length,
318 });
319 const abandon = new AbortController();
320 let timer: ReturnType<typeof setTimeout> | undefined;
321 let dropped: (() => void) | undefined;
322 // A gate that never answers is abandoned at the deadline: no verdict,
323 // and the next sweep forms the line again (#3430).
324 const cut = new Promise<LandVerdict>((resolve) => {
325 timer = setTimeout(() => {
326 log.warn("candidate abandoned", {
248327 repo: line.repoId,
249328 pulls,
250 waiting: line.waiting.length,
329 since: new Date(flight.since).toISOString(),
330 deadline: `${Math.round(deadlineMs / 1000)}s`,
251331 });
252 const abandon = new AbortController();
253 let timer: ReturnType<typeof setTimeout> | undefined;
254 // A gate that never answers is abandoned at the deadline: no
255 // verdict, and the next sweep forms the line again (#3430).
256 const deadline = new Promise<LandVerdict>((resolve) => {
257 timer = setTimeout(() => {
258 log.warn("candidate abandoned", {
259 repo: line.repoId,
260 pulls,
261 since: new Date(line.since).toISOString(),
262 deadline: `${Math.round(deadlineMs / 1000)}s`,
263 });
264 abandon.abort();
265 resolve({ lands: false, interrupted: true });
266 }, deadlineMs);
267 });
268 try {
269 const verdict = await Promise.race([gate(candidate, abandon.signal), deadline]);
270 log.info(`gated ${pullsWord(line.gating)}: ${outcome(verdict)}`, {
271 repo: line.repoId,
272 pulls,
273 took: `${Math.round((Date.now() - line.since) / 1000)}s`,
274 });
275 return verdict;
276 } finally {
277 clearTimeout(timer);
278 line.gating = 0;
279 line.since = 0;
280 }
332 abandon.abort();
333 resolve(UNDECIDED);
334 }, deadlineMs);
335 dropped = () => {
336 abandon.abort();
337 resolve(UNDECIDED);
281338 };
339 if (signal?.aborted === true) dropped();
340 else signal?.addEventListener("abort", dropped, { once: true });
341 });
342 try {
343 const verdict = await Promise.race([gate(candidate, abandon.signal), cut]);
344 log.info(`gated ${pullsWord(n)}: ${outcome(verdict)}`, {
345 repo: line.repoId,
346 pulls,
347 took: `${Math.round((Date.now() - flight.since) / 1000)}s`,
348 });
349 return verdict;
350 } finally {
351 clearTimeout(timer);
352 if (dropped !== undefined) signal?.removeEventListener("abort", dropped);
353 line.flights.delete(flight);
354 if (beyond) extra--;
355 }
356 };
357 }
358
359 /** A drain waiting on its gate, woken by pulls joining the line. */
360 const ARRIVED = Symbol("arrived");
361
362 /**
363 * Until the line is empty: keep up to `depth` candidates at the gate, the
364 * first on the base's tip and each next on the one before, and settle the
365 * first as its verdict comes. A batch is the front of the line that shares
366 * one place and one base branch; a sealed pull whose objects live
367 * elsewhere is a batch of its own, and is not stacked on another. A batch
368 * carries at most `MAX_CANDIDATE` pulls. A gate beyond the line's first
369 * starts only on a free check slot (`room`).
370 */
371 async function drain(line: Line, gate: GateCandidate, options: QueueOptions): Promise<void> {
372 line.draining = true;
373 const gated = counted(line, gate);
374 const depth = Math.max(1, Math.floor(options.depth ?? DEFAULT_DEPTH));
375 const room: Room = options.room ?? (() => 0);
376 const ahead: Flight[] = [];
377 try {
378 while (line.waiting.length > 0 || ahead.length > 0) {
379 while (ahead.length < depth && line.waiting.length > 0 && (ahead.length === 0 || room() > 0)) {
380 const flight = await launch(line, gated, ahead.at(-1));
381 if (flight === "apart") break;
382 if (flight !== undefined) ahead.push(flight);
383 }
384 const head = ahead[0];
385 if (head === undefined) continue;
386 const arrived =
387 ahead.length < depth && line.waiting.length === 0
388 ? new Promise<typeof ARRIVED>((resolve) => (line.wake = () => resolve(ARRIVED)))
389 : undefined;
390 const verdict = arrived === undefined ? await head.verdict : await Promise.race([head.verdict, arrived]);
391 line.wake = undefined;
392 if (verdict === ARRIVED) continue;
393 ahead.shift();
282394 let left: Waiting[];
283395 try {
284 left = await round(batch, counted);
396 left = await settleHead(head, verdict, ahead, gated, room);
285397 } catch (error) {
286398 log.warn("merge queue round failed", { error: String(error) });
287 for (const w of batch) w.settle(false);
288 left = [];
399 for (const w of head.steps) w.settle(false);
400 left = await discard(ahead);
289401 }
290402 // What waits for the next candidate goes back to the front.
291403 line.waiting = [...left, ...line.waiting];
@@ -296,13 +408,123 @@ async function drain(line: Line, gate: GateCandidate): Promise<void> {
296408 }
297409
298410 /**
299 * Build, gate and land one candidate of `batch`. `known` is a red verdict
300 * already read for a tree: a candidate that builds to that same tree is
301 * not gated again. Returns what waits for the next candidate.
411 * Take the next batch from the front of the line, build it — on `on`'s tip
412 * when there is a flight to stack on — and start its gate. `apart` when
413 * the front of the line cannot stack on `on`; undefined when nothing of
414 * the batch could be built.
415 */
416 async function launch(line: Line, gated: Gated, on: Flight | undefined): Promise<Flight | undefined | "apart"> {
417 const first = line.waiting[0]!.entry;
418 const branch = first.pull.baseBranch;
419 if (on !== undefined && (first.dir !== on.candidate.dir || branch !== on.branch)) return "apart";
420 const batch = line.waiting
421 .filter((w) => w.entry.dir === first.dir && w.entry.pull.baseBranch === branch)
422 .slice(0, MAX_CANDIDATE);
423 line.waiting = line.waiting.filter((w) => !batch.includes(w));
424 let candidate: Candidate | undefined;
425 try {
426 candidate = await build(batch, on?.candidate);
427 } catch (error) {
428 log.warn("merge queue round failed", { error: String(error) });
429 }
430 if (candidate === undefined || candidate.steps.length === 0) {
431 for (const w of batch) w.settle(false);
432 return undefined;
433 }
434 const steps = inCandidate(batch, candidate);
435 const abort = new AbortController();
436 const verdict = gated(candidate, abort.signal).catch((error: unknown) => {
437 log.warn("merge queue round failed", { error: String(error) });
438 return UNDECIDED;
439 });
440 return { steps, candidate, speculative: on !== undefined, branch, verdict, abort };
441 }
442
443 /**
444 * The pulls of `batch` the candidate carries. The rest — a conflict with
445 * the base or an earlier step, or nothing left to land once the steps
446 * before it did — are not this candidate's; the next sweep brings them up
447 * to date or closes them as carried.
448 */
449 function inCandidate(batch: readonly Waiting[], candidate: Candidate): Waiting[] {
450 const inIt = new Set(candidate.steps.map((s) => s.entry));
451 for (const w of batch) if (!inIt.has(w.entry)) w.settle(false);
452 return batch.filter((w) => inIt.has(w.entry));
453 }
454
455 /**
456 * The first flight's verdict, acted on: green lands it and keeps the
457 * flights behind it, anything else discards them back to the line. A
458 * speculative flight is settled on the base's tip only when that tip is
459 * the tree it was built on — the flight before it landed whole. Returns
460 * what waits for the next candidate.
461 */
462 async function settleHead(
463 head: Flight,
464 verdict: LandVerdict,
465 ahead: Flight[],
466 gated: Gated,
467 room: Room,
468 ): Promise<Waiting[]> {
469 let candidate = head.candidate;
470 if (head.speculative) {
471 const tip = await onTip(candidate, head.branch);
472 if (tip === undefined) {
473 // What it was stacked on did not land as gated: formed again on main.
474 return [...head.steps, ...(await discard(ahead))];
475 }
476 candidate = { ...candidate, baseOid: tip };
477 }
478 if (verdict.lands) {
479 if (await land(candidate, head.steps)) return [];
480 return discard(ahead);
481 }
482 const behind = await discard(ahead);
483 // The gate was cut off (the forge stopping, the Job lost): nothing is
484 // decided and no one is handed back; the next sweep admits them again (#3155).
485 if ("interrupted" in verdict) {
486 for (const w of head.steps) w.settle(false);
487 return behind;
488 }
489 return [...(await judged(candidate, head.steps, verdict, gated, room)), ...behind];
490 }
491
492 /**
493 * The base's tip, when its tree is the one `candidate` was built on:
494 * undefined when the base moved elsewhere.
495 */
496 async function onTip(candidate: Candidate, branch: string): Promise<string | undefined> {
497 const tip = await resolveOid(candidate.dir, `refs/heads/${branch}`);
498 if (tip === undefined) return undefined;
499 if (tip === candidate.baseOid) return tip;
500 return (await treeOf(candidate.dir, tip)) === candidate.baseTreeOid ? tip : undefined;
501 }
502
503 /**
504 * Stop the flights in `ahead` — what they were stacked on did not land —
505 * and hand their pulls back for the next candidate, in order.
506 */
507 async function discard(ahead: Flight[]): Promise<Waiting[]> {
508 const gone = ahead.splice(0);
509 if (gone.length === 0) return [];
510 for (const f of gone) f.abort.abort();
511 await Promise.all(gone.map((f) => f.verdict));
512 log.info("speculative candidates re-formed", {
513 pulls: gone.flatMap((f) => f.steps.map((w) => `#${w.entry.pull.number}`)).join(","),
514 });
515 return gone.flatMap((f) => [...f.steps]);
516 }
517
518 /**
519 * Build, gate and land one candidate of `batch` on the base's tip. `known`
520 * is a red verdict already read for a tree: a candidate that builds to
521 * that same tree is not gated again. Returns what waits for the next
522 * candidate.
302523 */
303524 async function round(
304525 batch: readonly Waiting[],
305 gate: Gated,
526 gated: Gated,
527 room: Room,
306528 known?: { treeOid: string; red: Red },
307529 ): Promise<Waiting[]> {
308530 const candidate = await build(batch);
@@ -310,63 +532,90 @@ async function round(
310532 for (const w of batch) w.settle(false);
311533 return [];
312534 }
313 const inIt = new Set(candidate.steps.map((s) => s.entry));
314 // Conflicts with the base or an earlier step, or nothing left to land
315 // once the steps before it did: not this candidate's; the next sweep
316 // brings it up to date or closes it as carried.
317 for (const w of batch) if (!inIt.has(w.entry)) w.settle(false);
318 const steps = batch.filter((w) => inIt.has(w.entry));
535 const steps = inCandidate(batch, candidate);
319536 const verdict: LandVerdict =
320 known !== undefined && known.treeOid === candidate.treeOid
321 ? known.red
322 : await gate(candidate);
537 known !== undefined && known.treeOid === candidate.treeOid ? known.red : await gated(candidate);
323538 if (verdict.lands) {
324539 await land(candidate, steps);
325540 return [];
326541 }
327 // The gate was cut off (the forge stopping, the Job lost): nothing is
328 // decided and no one is handed back; the next sweep admits them again (#3155).
329542 if ("interrupted" in verdict) {
330543 for (const w of steps) w.settle(false);
331544 return [];
332545 }
546 return judged(candidate, steps, verdict, gated, room);
547 }
548
549 /**
550 * A red candidate: one step is the culprit, handed back on the base it was
551 * red on. More are searched (#4053): the candidate is built again on the
552 * base's tip and `k` of its prefixes are gated at once — one, the front
553 * half, plus one per free check slot, evenly cut. The longest green prefix
554 * before the first red one lands; the culprit is among the steps up to the
555 * first red prefix (or, all green, the rest of the candidate, whose whole
556 * tree is the red one), searched again; the steps after wait for the next
557 * candidate.
558 */
559 async function judged(
560 candidate: Candidate,
561 steps: readonly Waiting[],
562 red: Red,
563 gated: Gated,
564 room: Room,
565 ): Promise<Waiting[]> {
333566 if (steps.length === 1) {
334 steps[0]!.entry.handBack(verdict, candidate.baseOid);
567 steps[0]!.entry.handBack(red, candidate.baseOid);
335568 steps[0]!.settle(false);
336569 return [];
337570 }
338 const red = { treeOid: candidate.treeOid, red: verdict };
339 const half = Math.ceil(steps.length / 2);
340 const front = steps.slice(0, half);
341 const rest = steps.slice(half);
342 const part = await build(front);
343 if (part === undefined || part.steps.length < front.length) {
571 const n = steps.length;
572 const k = Math.min(n - 1, 1 + Math.max(0, room()));
573 const cuts = [...new Set(Array.from({ length: k }, (_, j) => Math.ceil(((j + 1) * n) / (k + 1))))].filter(
574 (c) => c >= 1 && c < n,
575 );
576 const last = cuts.at(-1)!;
577 const whole = await build(steps);
578 if (whole === undefined || whole.steps.slice(0, last).some((s, i) => s.entry !== steps[i]!.entry) || whole.steps.length < last) {
344579 // The base moved under the search: nothing here is decided, and the
345580 // next sweep admits these again onto the base as it is now.
346581 for (const w of steps) w.settle(false);
347582 return [];
348583 }
349 const frontVerdict = await gate(part);
350 if (frontVerdict.lands) {
351 await land(part, front);
352 // The front was green, the whole red: the culprit is in the rest, and
353 // the rest on the new base is the red tree again.
354 return round(rest, gate, red);
584 const prefix = (c: number): Candidate => {
585 const end = whole.steps[c - 1]!;
586 return { ...whole, steps: whole.steps.slice(0, c), tipOid: end.commitOid, treeOid: end.treeOid };
587 };
588 const verdicts = await Promise.all(cuts.map((c) => gated(prefix(c))));
589 let green = 0;
590 let stop: Exclude<LandVerdict, { lands: true }> | undefined;
591 let stopAt = n;
592 for (const [j, c] of cuts.entries()) {
593 const v = verdicts[j]!;
594 if (v.lands) {
595 green = c;
596 continue;
597 }
598 stop = v;
599 stopAt = c;
600 break;
355601 }
356 if ("interrupted" in frontVerdict) {
357 for (const w of steps) w.settle(false);
602 if (green > 0) await land(prefix(green), steps.slice(0, green));
603 if (stop !== undefined && "interrupted" in stop) {
604 for (const w of steps.slice(green)) w.settle(false);
358605 return [];
359606 }
360 const left = await round(front, gate, { treeOid: part.treeOid, red: frontVerdict });
361 return [...left, ...rest];
607 const known = stop === undefined ? { treeOid: candidate.treeOid, red } : { treeOid: prefix(stopAt).treeOid, red: stop };
608 const left = await round(steps.slice(green, stopAt), gated, room, known);
609 return [...left, ...steps.slice(stopAt)];
362610 }
363611
364612 /**
365 * Land each step in order; a step that does not land leaves the rest for
366 * the line. The last step that did land is released, so a prefix that
367 * reached the base is never left without one.
613 * Land each step in order, from the candidate's base; a step that does not
614 * land leaves the rest for the line. The last step that did land is
615 * released, so a prefix that reached the base is never left without one.
616 * Whether every step landed.
368617 */
369 async function land(candidate: Candidate, steps: readonly Waiting[]): Promise<void> {
618 async function land(candidate: Candidate, steps: readonly Waiting[]): Promise<boolean> {
370619 let at = candidate.baseOid;
371620 let landed: Queued | undefined;
372621 try {
@@ -380,31 +629,34 @@ async function land(candidate: Candidate, steps: readonly Waiting[]): Promise<vo
380629 of: candidate.steps.length,
381630 });
382631 for (const rest of steps.slice(i)) rest.settle(false);
383 return;
632 return false;
384633 }
385634 w.settle(true);
386635 at = tip;
387636 landed = step.entry;
388637 }
638 return true;
389639 } finally {
390640 landed?.release(at);
391641 }
392642 }
393643
394644 /**
395 * The candidate of `batch` on its base's tip now: each pull's squash on the
396 * step before it. A pull that conflicts, or whose squash changes nothing,
397 * is left out. Undefined when the base cannot be read.
645 * The candidate of `batch` on its base's tip now — or on `on`'s tip, to
646 * gate it ahead of `on` landing: each pull's squash on the step before it.
647 * A pull that conflicts, or whose squash changes nothing, is left out.
648 * Undefined when the base cannot be read.
398649 */
399 async function build(batch: readonly Waiting[]): Promise<Candidate | undefined> {
650 async function build(batch: readonly Waiting[], on?: Candidate): Promise<Candidate | undefined> {
400651 const first = batch[0]?.entry;
401652 if (first === undefined) return undefined;
402653 const dir = first.dir;
403 const baseOid = await resolveOid(dir, `refs/heads/${first.pull.baseBranch}`);
654 const baseOid = on?.tipOid ?? (await resolveOid(dir, `refs/heads/${first.pull.baseBranch}`));
404655 if (baseOid === undefined) return undefined;
405656 let tipOid = baseOid;
406 let treeOid = await treeOf(dir, baseOid);
407 if (treeOid === undefined) return undefined;
657 const baseTreeOid = on?.treeOid ?? (await treeOf(dir, baseOid));
658 if (baseTreeOid === undefined) return undefined;
659 let treeOid = baseTreeOid;
408660 const steps: Step[] = [];
409661 for (const { entry } of batch) {
410662 const merged = await mergeable(dir, tipOid, entry.pull.headOid);
@@ -414,9 +666,10 @@ async function build(batch: readonly Waiting[]): Promise<Candidate | undefined>
414666 { cwd: dir, env: IDENTITY },
415667 );
416668 if (!commit.ok) continue;
417 steps.push({ entry, baseOid: tipOid, treeOid: merged.value.treeOid });
418 tipOid = commit.value.trim();
669 const commitOid = commit.value.trim();
670 steps.push({ entry, baseOid: tipOid, treeOid: merged.value.treeOid, commitOid });
671 tipOid = commitOid;
419672 treeOid = merged.value.treeOid;
420673 }
421 return { dir, baseOid, steps, tipOid, treeOid };
674 return { dir, baseOid, baseTreeOid, steps, tipOid, treeOid };
422675 }
src/checks/service.ts +16 −1
@@ -84,6 +84,7 @@ import { summariseCheckFailure } from "./summary.ts";
8484 import type { Result } from "../kernel/result.ts";
8585 import { placementFromEnv, type Placement } from "./placement.ts";
8686 import { JOB_VANISHED, runJobStep } from "./job_step.ts";
87 import { extraGates } from "./queue.ts";
8788 import type { JobOptions } from "../cluster/jobs.ts";
8889 import type { CheckDetail } from "../core/runs.ts";
8990
@@ -158,10 +159,22 @@ export interface ChecksHandle {
158159 * is false (a person's Ctrl-C): then they are cut off and queued again now.
159160 */
160161 stop(options?: { drain?: boolean }): Promise<void>;
162 /**
163 * Check slots free now for a merge-queue gate beyond a line's first
164 * (#4053): what neither a running check nor such a gate holds, less
165 * `PULL_RESERVE` kept for pull checks.
166 */
167 spareSlots(): number;
161168 /** The gate checks take their turn at; a hosted child's checks share it. */
162169 readonly lane: Gate;
163170 }
164171
172 /**
173 * Check slots the merge queue's extra gates never take (#4053): pull checks
174 * always have at least these, however long the queue.
175 */
176 export const PULL_RESERVE = 2;
177
165178 /**
166179 * How long main's check after a land waits to start (#4061). The merge queue
167180 * lands a candidate's steps seconds apart, and only its tip was gated: a
@@ -419,7 +432,8 @@ export function startChecks(app: App, options: ChecksOptions): ChecksHandle {
419432 if (stopped) return;
420433 const limit = capacity();
421434 if (options.gate === undefined) gate.resize(limit);
422 while (inFlight.size < limit) {
435 // A merge-queue gate beyond a line's first holds a slot (#4053).
436 while (inFlight.size < limit - extraGates()) {
423437 // A pull's check waits while its main is red (`checks/red_main.ts`).
424438 const queued = unheld(
425439 listActive(app.db, "check").filter(
@@ -1180,6 +1194,7 @@ export function startChecks(app: App, options: ChecksOptions): ChecksHandle {
11801194 gates.delete(controller);
11811195 }
11821196 },
1197 spareSlots: () => Math.max(0, capacity() - PULL_RESERVE - inFlight.size - extraGates()),
11831198 lane: gate,
11841199 stop: async (stopOptions) => {
11851200 stopped = true;
src/core/capacity.ts +13
@@ -19,6 +19,7 @@ export type CapacityKey =
1919 | "crew.concurrency"
2020 | "checks.concurrency"
2121 | "checks.shards"
22 | "queue.depth"
2223 | "children.max"
2324 | "cells.egress"
2425 | "guard_new_accounts"
@@ -64,6 +65,13 @@ export const CAPACITY: Record<CapacityKey, Definition> = {
6465 sized: (cpus) => shardsForCpus(cpus),
6566 help: "How many parallel shards a check's test run uses (handed to checks and crew shells as FORGE_CHECK_SHARDS). Empty: the pod's CPUs, at most 4 — and, in a check, no more than its cell holds.",
6667 },
68 "queue.depth": {
69 label: "Merge queue candidates at once",
70 env: "FORGE_QUEUE_DEPTH",
71 kind: { type: "count", min: 1, max: 4 },
72 fallback: "2",
73 help: "How many merge-queue candidates of one repository gate at once, each on top of the one before: while one gates on main the next gates on it, and lands right after it if both are green. 1 gates one candidate at a time. A gate beyond the first takes a free check slot only, never one of the two kept for pull checks.",
74 },
6775 "children.max": {
6876 label: "Forges this one may send",
6977 env: "FORGE_CHILDREN_MAX",
@@ -305,6 +313,11 @@ export function egressSetting(db: Database): "enforce" | "report" {
305313 return capacityValue(db, "cells.egress") === "report" ? "report" : "enforce";
306314 }
307315
316 /** How many merge-queue candidates of one repository gate at once (#4053). */
317 export function queueDepth(db: Database): number {
318 return Number(capacityValue(db, "queue.depth"));
319 }
320
308321 /** Whether the platform edge's identity is read at all; on every request. */
309322 export function edgeSigninOn(db: Database): boolean {
310323 return capacityValue(db, "signin.edge") !== "off";
src/crew/autonomy.ts +41 −5
@@ -23,7 +23,9 @@ import {
2323 import type { AccessError } from "../actions/access.ts";
2424 import { gateCandidate, releaseOnLand } from "../checks/landing.ts";
2525 import { addLearned } from "../checks/learned.ts";
26 import { enqueue, type GateCandidate, type Queued, queueWords } from "../checks/queue.ts";
26 import { enqueue, type GateCandidate, type Queued, type QueueOptions, queueWords } from "../checks/queue.ts";
27 import { checksFor } from "../checks/service.ts";
28 import { queueDepth } from "../core/capacity.ts";
2729 import { mergeable, resolveOid, treeOf } from "../git/merge.ts";
2830 import { repoDir } from "../git/repos.ts";
2931 import { CLOSED_NOTE } from "../core/done_on_main.ts";
@@ -53,6 +55,7 @@ import { wallOf, wallSentence } from "../contract/policy.ts";
5355 import {
5456 getIssue,
5557 getIssueById,
58 LIST_LIMIT_MAX,
5659 listComments,
5760 MAX_TITLE,
5861 type Issue,
@@ -1039,11 +1042,10 @@ export async function sweepGreen(app: App, repo: Repo): Promise<number> {
10391042 if (repo.autonomy !== "autonomous") return 0;
10401043 // Admitted one by one, queued together, oldest first: the green pulls of
10411044 // one sweep are one candidate and one gate (#2036).
1042 const open = listPulls(app.db, repo.id, { sealed: true, state: "open", limit: 50 });
10431045 const entries: Queued[] = [];
10441046 // Why each approved pull waits, for the tick line (#3055).
10451047 const waits: Wait[] = [];
1046 for (const pull of open.sort((a, b) => a.number - b.number)) {
1048 for (const pull of openPulls(app, repo)) {
10471049 const admitted = await admit(app, repo, pull);
10481050 if (admitted.ok) entries.push(admitted.value);
10491051 const seen = reviewSummary(app.db, pull.id, pull.headOid);
@@ -1052,10 +1054,33 @@ export async function sweepGreen(app: App, repo: Repo): Promise<number> {
10521054 }
10531055 noteWaits(repo.id, tallyWaits(waits));
10541056 if (entries.length === 0) return 0;
1055 const landed = await enqueue(repo.id, entries, gateFor(app, repo));
1057 const landed = await enqueue(repo.id, entries, gateFor(app, repo), queueOptions(app));
10561058 return landed.filter(Boolean).length;
10571059 }
10581060
1061 /**
1062 * Every open pull of the repository, oldest first. All of them, page by
1063 * page: a sweep that read only the newest page never queued an approved
1064 * pull older than it, which then joined the line alone, on its own event,
1065 * one gate at a time (#4053).
1066 */
1067 function openPulls(app: App, repo: Repo): Pull[] {
1068 const all: Pull[] = [];
1069 let cursor: number | undefined;
1070 for (;;) {
1071 const page = listPulls(app.db, repo.id, {
1072 sealed: true,
1073 state: "open",
1074 limit: LIST_LIMIT_MAX,
1075 ...(cursor === undefined ? {} : { cursor }),
1076 });
1077 all.push(...page);
1078 if (page.length < LIST_LIMIT_MAX) break;
1079 cursor = page.at(-1)!.number;
1080 }
1081 return all.sort((a, b) => a.number - b.number);
1082 }
1083
10591084 /** The sweep each repository has in flight, by repository id. */
10601085 const sweeping = new Map<string, Promise<number>>();
10611086
@@ -1644,7 +1669,7 @@ export async function landIfGreen(
16441669 ): Promise<boolean> {
16451670 const admitted = await admit(app, repo, pull);
16461671 if (!admitted.ok) return false;
1647 const [landed] = await enqueue(repo.id, [admitted.value], gateFor(app, repo));
1672 const [landed] = await enqueue(repo.id, [admitted.value], gateFor(app, repo), queueOptions(app));
16481673 return landed === true;
16491674 }
16501675
@@ -1857,6 +1882,17 @@ export function regateDue(red: Decision, now: number): boolean {
18571882 return now - Date.parse(red.createdAt) >= REGATE_RED_MS;
18581883 }
18591884
1885 /**
1886 * How the merge queue may use the forge's check slots (#4053): its depth as
1887 * an admin set it, and the slots the checks service has spare.
1888 */
1889 function queueOptions(app: App): QueueOptions {
1890 return {
1891 depth: queueDepth(app.db),
1892 room: () => checksFor(app)?.spareSlots?.() ?? 0,
1893 };
1894 }
1895
18601896 /** The queue's gate: the full tier on the candidate's tip, asked as its first pull. */
18611897 function gateFor(app: App, repo: Repo): GateCandidate {
18621898 return (candidate, signal) =>
src/flow/queue.ts +38 −5
@@ -1,9 +1,10 @@
11 /**
22 * The merge queue as /flow shows it (#2036): per repository, the pulls
3 * queued for the next candidate, the pulls in the candidate at the gate,
4 * and the pulls landed on the default branch in the last hour. Queued and
5 * gating are this process's line (`checks/queue.ts`); landed is the rows.
6 * A repository with none of the three is not listed.
3 * queued for the next candidate, the pulls in candidates at the gate, the
4 * gates in flight (#4053), the pulls landed on the default branch in the
5 * last hour, and their median minutes from approval to landing. Queued,
6 * gating and gates are this process's line (`checks/queue.ts`); landed is
7 * the rows. A repository with nothing queued, gating or landed is not listed.
78 */
89
910 import type { Database } from "bun:sqlite";
@@ -14,12 +15,19 @@ export interface QueueRow {
1415 readonly repo: string;
1516 readonly queued: number;
1617 readonly gating: number;
18 /** Gates in flight: more than one when the queue gates in parallel. */
19 readonly gates: number;
1720 readonly landedHour: number;
21 /**
22 * Median whole minutes from a landed pull's last approval to its landing,
23 * over the last hour's; null when none of them was approved.
24 */
25 readonly approvedToLanded: number | null;
1826 }
1927
2028 const HOUR_MS = 60 * 60 * 1000;
2129
22 /** Two queries, however many repositories have a line. */
30 /** Three queries, however many repositories have a line. */
2331 export function readQueues(db: Database, now: Date): QueueRow[] {
2432 const since = new Date(now.getTime() - HOUR_MS).toISOString();
2533 const landed = db
@@ -32,6 +40,20 @@ export function readQueues(db: Database, now: Date): QueueRow[] {
3240 GROUP BY r.id`,
3341 )
3442 .all(since, now.toISOString());
43 const waits = new Map<string, number[]>();
44 for (const w of db
45 .query<{ id: string; ms: number }, [string, string]>(
46 `SELECT i.repo_id AS id,
47 (julianday(p.merged_at) - julianday(MAX(v.created_at))) * 86400000 AS ms
48 FROM pull p JOIN issue i ON i.id = p.issue_id
49 JOIN repo r ON r.id = i.repo_id
50 JOIN review v ON v.issue_id = p.issue_id AND v.state = 'approved' AND v.created_at <= p.merged_at
51 WHERE p.base_branch = r.default_branch
52 AND p.merged_at >= ?1 AND p.merged_at <= ?2
53 GROUP BY p.issue_id`,
54 )
55 .all(since, now.toISOString()))
56 waits.set(w.id, [...(waits.get(w.id) ?? []), w.ms]);
3557 const lines = new Map(queueStatus().map((s) => [s.repoId, s]));
3658 const named = new Map(landed.map((l) => [l.id, l.full]));
3759 const unnamed = [...lines.keys()].filter((id) => !named.has(id));
@@ -50,7 +72,18 @@ export function readQueues(db: Database, now: Date): QueueRow[] {
5072 repo,
5173 queued: lines.get(id)?.queued ?? 0,
5274 gating: lines.get(id)?.gating ?? 0,
75 gates: lines.get(id)?.gates ?? 0,
5376 landedHour: count.get(id) ?? 0,
77 approvedToLanded: medianMinutes(waits.get(id) ?? []),
5478 }))
5579 .sort((a, b) => b.gating + b.queued - (a.gating + a.queued) || a.repo.localeCompare(b.repo));
5680 }
81
82 /** The median of `ms`, in whole minutes; null for none. */
83 function medianMinutes(ms: readonly number[]): number | null {
84 if (ms.length === 0) return null;
85 const sorted = [...ms].sort((a, b) => a - b);
86 const mid = Math.floor(sorted.length / 2);
87 const median = sorted.length % 2 === 1 ? sorted[mid]! : (sorted[mid - 1]! + sorted[mid]!) / 2;
88 return Math.round(median / 60_000);
89 }
src/tui/screens/flow.ts +4 −3
@@ -4,7 +4,8 @@
44 * stage of what left it in the last hour and in 24 hours — the numbers of
55 * `flow/pipeline.ts` and `flow/throughput.ts`, the same the page and the
66 * MCP `flow` tool read. Under them, each repository's merge queue (#2036):
7 * queued, gating, landed in the last hour (`flow/queue.ts`); then the
7 * queued, gating, gates in flight, landed in the last hour and the median
8 * minutes from approval to landing (`flow/queue.ts`); then the
89 * conflicted pulls, and the pulls with changes requested waiting on a
910 * builder and on a reviewer (`flow/requested.ts`, #2172); then each
1011 * autonomous repository's ready list, startable beside the count per
@@ -60,12 +61,12 @@ export function flowLines(
6061 `${pad(STAGE_LABEL[s.key], 12)} ${pad(String(s.lastHour), 9, "right")} ${pad(String(s.day), 5, "right")} ${s.label}`,
6162 ),
6263 "",
63 `${pad("Merge queue", 30)} ${pad("queued", 6, "right")} ${pad("gating", 6, "right")} ${pad("landed 1h", 9, "right")}`,
64 `${pad("Merge queue", 15)} ${pad("queued", 6, "right")} ${pad("gating", 6, "right")} ${pad("gates", 5, "right")} ${pad("landed 1h", 9, "right")} ${pad("to land, min", 12, "right")}`,
6465 ...(queues.length === 0
6566 ? ["Nothing queued, gating or landed in the last hour."]
6667 : queues.map(
6768 (q) =>
68 `${pad(truncate(q.repo, 30), 30)} ${pad(String(q.queued), 6, "right")} ${pad(String(q.gating), 6, "right")} ${pad(String(q.landedHour), 9, "right")}`,
69 `${pad(truncate(q.repo, 15), 15)} ${pad(String(q.queued), 6, "right")} ${pad(String(q.gating), 6, "right")} ${pad(String(q.gates), 5, "right")} ${pad(String(q.landedHour), 9, "right")} ${pad(q.approvedToLanded === null ? "—" : String(q.approvedToLanded), 12, "right")}`,
6970 )),
7071 ...(conflicted === null ? [] : ["", ...wrap(conflicted, inner)]),
7172 ...waits,
src/web/pages/flow-panel.ts +7 −1
@@ -172,7 +172,9 @@ function queueCard(rows: readonly QueueRow[]): Html {
172172 html`<h2>Merge queue</h2>
173173 <p class="muted small">
174174 Green, approved pulls join their repository's queue; each candidate
175 is gated once and lands whole, or is bisected to the red one.
175 is gated once and lands whole, or is searched for the red one. Where
176 check slots are free the next candidate gates on top of this one, and
177 a red one's halves gate at once.
176178 </p>
177179 ${rows.length === 0
178180 ? html`<p class="muted">Nothing queued, gating or landed in the last hour.</p>`
@@ -183,7 +185,9 @@ function queueCard(rows: readonly QueueRow[]): Html {
183185 <th scope="col">Repository</th>
184186 <th scope="col">Queued</th>
185187 <th scope="col">Gating</th>
188 <th scope="col">Gates</th>
186189 <th scope="col">Landed, last hour</th>
190 <th scope="col">Approved to landed</th>
187191 </tr>
188192 </thead>
189193 <tbody>
@@ -193,7 +197,9 @@ function queueCard(rows: readonly QueueRow[]): Html {
193197 <th scope="row"><a href="/${r.repo}/pulls">${r.repo}</a></th>
194198 <td class="mono" data-queued>${String(r.queued)}</td>
195199 <td class="mono" data-gating>${String(r.gating)}</td>
200 <td class="mono" data-gates>${String(r.gates)}</td>
196201 <td class="mono" data-landed>${String(r.landedHour)}</td>
202 <td class="mono" data-approved-to-landed>${r.approvedToLanded === null ? "—" : `${r.approvedToLanded} min`}</td>
197203 </tr>`,
198204 )}
199205 </tbody>
test/merge-queue-parallel.test.ts added +313
@@ -0,0 +1,313 @@
1 /**
2 * The merge queue gates in parallel where check slots are free (#4053).
3 * While one candidate gates on main the next gates on top of it, and lands
4 * right after it; a red candidate's prefixes gate at once, so the culprit
5 * is found in one gate-time, not one per halving. The gate here is a
6 * stand-in the test holds open: it lets a whole wave go only once every
7 * gate the wave should hold has started, so "at once" is counted, never
8 * timed. It goes red on any tree holding a `red-*` file.
9 */
10 import { afterAll, beforeAll, describe, expect, test } from "bun:test";
11 import { mkdtemp, rm } from "node:fs/promises";
12 import { tmpdir } from "node:os";
13 import { join } from "node:path";
14 import { createApp, type App } from "../src/app.ts";
15 import { openPull } from "../src/actions/pulls.ts";
16 import { commitToBranch, createRepo } from "../src/actions/repos.ts";
17 import { enqueue, extraGates, gateRuns, queueStatus, queueWords } from "../src/checks/queue.ts";
18 import { PULL_RESERVE, provideChecksForTest, startChecks, type ChecksHandle } from "../src/checks/service.ts";
19 import { latestDecision } from "../src/core/decisions.ts";
20 import { getPull, type Pull } from "../src/core/pulls.ts";
21 import { getRepoById, type Repo } from "../src/core/repos.ts";
22 import { submitReview } from "../src/core/reviews.ts";
23 import { claimRun, createRun, finishRun } from "../src/core/runs.ts";
24 import { createUser, type User } from "../src/core/users.ts";
25 import { sweepGreen } from "../src/crew/autonomy.ts";
26 import { readQueues } from "../src/flow/queue.ts";
27 import { repoDir } from "../src/git/repos.ts";
28 import { flowLines } from "../src/tui/screens/flow.ts";
29 import { serve, type Servers } from "../src/http/server.ts";
30
31 /** Check slots the stand-in service has. */
32 const CAPACITY = 12;
33
34 describe("the merge queue in parallel", () => {
35 let dataDir: string;
36 let app: App;
37 let alice: User;
38 let servers: Servers;
39 /** Gates started and not yet let go. */
40 let pending: { commit: string; go: () => void }[] = [];
41 /** Every gate, with its verdict and the wave it was let go in. */
42 const gated: { repo: string; commit: string; ok: boolean; wave: number }[] = [];
43 let wave = 0;
44
45 const git = (repo: Repo, ...args: string[]) =>
46 Bun.spawnSync(["git", ...args], { cwd: repoDir(app.reposRoot, repo.ownerName, repo.name) })
47 .stdout.toString()
48 .trim();
49
50 beforeAll(async () => {
51 dataDir = await mkdtemp(join(tmpdir(), "forge-merge-queue-parallel-"));
52 app = createApp({ dataDir, port: 0 });
53 const a = await createUser(app.db, {
54 origin: "seed",
55 trusted: true,
56 username: "alice",
57 email: "[email protected]",
58 password: "a-good-passphrase",
59 adminIfFirst: true,
60 });
61 if (!a.ok) throw new Error(a.error);
62 alice = a.value;
63 provideChecksForTest(app, {
64 gate: async (repo: Repo, _pull: Pull, commitOid: string) => {
65 await new Promise<void>((go) => pending.push({ commit: commitOid, go }));
66 const files = git(repo, "ls-tree", "-r", "--name-only", commitOid).split("\n");
67 const ok = !files.some((f) => f.startsWith("red-"));
68 gated.push({ repo: repo.id, commit: commitOid, ok, wave });
69 return ok ? { ok, error: "", lines: [] } : { ok, error: "red", lines: ["test/red.test.ts:", "(fail) goes red [1.00ms]"] };
70 },
71 // Nothing else runs: every slot but the pull checks' reserve is free.
72 spareSlots: () => Math.max(0, CAPACITY - PULL_RESERVE - extraGates()),
73 } as unknown as ChecksHandle);
74 servers = await serve(app);
75 });
76
77 afterAll(async () => {
78 servers.stop();
79 await app.close?.();
80 await rm(dataDir, { recursive: true, force: true });
81 });
82
83 async function autonomousRepo(name: string): Promise<Repo> {
84 const made = await createRepo(app, alice, { name, visibility: "private" });
85 if (!made.ok) throw new Error(String(made.error));
86 const seeded = await commitToBranch(app, alice, made.value, {
87 message: "Start",
88 files: [
89 { path: "README.md", content: `# ${name}\n` },
90 { path: "forge.json", content: JSON.stringify({ check: { run: "true", fast: "true", full: "true" } }) },
91 ],
92 });
93 if (!seeded.ok) throw new Error(JSON.stringify(seeded.error));
94 app.db.query("UPDATE repo SET autonomy = 'autonomous', autonomy_by = ?2 WHERE id = ?1").run(made.value.id, alice.id);
95 return getRepoById(app.db, made.value.id)!;
96 }
97
98 async function greenPull(repo: Repo, path: string): Promise<Pull> {
99 const pushed = await commitToBranch(app, alice, repo, {
100 branch: `feature/${path}`,
101 base: "main",
102 message: path,
103 files: [{ path, content: `export const n = "${path}";\n` }],
104 });
105 if (!pushed.ok) throw new Error(JSON.stringify(pushed.error));
106 const opened = await openPull(app, alice, repo, { title: path, baseBranch: "main", headBranch: `feature/${path}` });
107 if (!opened.ok) throw new Error(String(opened.error));
108 const pull = getPull(app.db, repo.id, opened.value.number)!;
109 const check = createRun(app.db, {
110 repoId: repo.id,
111 kind: "check",
112 issueId: pull.id,
113 env: `pull-${pull.number}`,
114 commitOid: pull.headOid,
115 reason: "pull",
116 });
117 claimRun(app.db, check.id);
118 finishRun(app.db, check.id, { state: "ok", summary: "check passed" });
119 return pull;
120 }
121
122 /** Wait until `n` gates are open at once, then let them all go: one wave. */
123 async function letGo(n: number): Promise<void> {
124 for (let i = 0; i < 2000 && pending.length < n; i++) await Bun.sleep(5);
125 expect(pending).toHaveLength(n);
126 wave++;
127 const going = pending;
128 pending = [];
129 for (const p of going) p.go();
130 }
131
132 const merged = (repo: Repo, pull: Pull) => getPull(app.db, repo.id, pull.number)!.mergedAt !== null;
133 const tree = (repo: Repo, rev: string) => git(repo, "rev-parse", `${rev}^{tree}`);
134
135 test("sixteen green pulls: two candidates gate at once, the second on the first, and both land in one gate-time", async () => {
136 const repo = await autonomousRepo("sixteen-green");
137 const pulls = [];
138 for (let i = 0; i < 16; i++) pulls.push(await greenPull(repo, `p${i}.ts`));
139 const sweep = sweepGreen(app, repo);
140 for (let i = 0; i < 2000 && pending.length < 2; i++) await Bun.sleep(5);
141 expect(queueStatus().find((s) => s.repoId === repo.id)).toMatchObject({ gating: 16, gates: 2, queued: 0 });
142 expect(queueWords(repo.id)).toMatch(/^gating 16 pulls since \d\d:\d\d in 2 gates$/);
143 await letGo(2);
144 expect(await sweep).toBe(16);
145 expect(gateRuns(repo.id)).toBe(2);
146 for (const pull of pulls) expect(merged(repo, pull)).toBe(true);
147 // Main is exactly the tree the second candidate was gated on.
148 const mine = gated.filter((g) => g.repo === repo.id);
149 expect(mine.every((g) => g.ok && g.wave === wave)).toBe(true);
150 expect(mine.map((g) => tree(repo, g.commit))).toContain(tree(repo, "refs/heads/main"));
151 expect(git(repo, "rev-list", "--count", "refs/heads/main")).toBe("17");
152 });
153
154 test("ten pulls, one red: the culprit is found in one wave of prefixes, and the nine land by the third gate-time", async () => {
155 const repo = await autonomousRepo("ten-one-red");
156 const pulls = [];
157 for (let i = 0; i < 10; i++) pulls.push(await greenPull(repo, i === 4 ? `red-${i}.ts` : `p${i}.ts`));
158 const first = wave;
159 const sweep = sweepGreen(app, repo);
160 // Wave one: eight on main, the other two on top of them.
161 await letGo(2);
162 // Wave two: the eight are red; the two on top are re-formed, and seven
163 // prefixes of the eight gate at once: four green, the fifth red.
164 await letGo(7);
165 // Wave three: the culprit is handed back without a gate of its own;
166 // the three after it and the two re-formed are one candidate.
167 await letGo(1);
168 expect(await sweep).toBe(9);
169 expect(wave - first).toBe(3);
170 const red = pulls[4]!;
171 for (const pull of pulls) expect(merged(repo, pull)).toBe(pull !== red);
172 expect(latestDecision(app.db, red.id, "land")?.answer).toBe("red");
173 expect(git(repo, "ls-tree", "--name-only", "refs/heads/main")).not.toContain("red-4.ts");
174 const green = gated.filter((g) => g.repo === repo.id && g.ok).map((g) => tree(repo, g.commit));
175 expect(green).toContain(tree(repo, "refs/heads/main"));
176 });
177
178 test("a red second candidate does not hold back the green first, and is searched on main once the first landed", async () => {
179 const repo = await autonomousRepo("red-behind");
180 const pulls = [];
181 for (let i = 0; i < 10; i++) pulls.push(await greenPull(repo, i === 9 ? `red-${i}.ts` : `p${i}.ts`));
182 const sweep = sweepGreen(app, repo);
183 await letGo(2);
184 // The first eight landed; the two behind are red on them, so one is
185 // gated alone (the only prefix of two), and the culprit handed back.
186 await letGo(1);
187 expect(await sweep).toBe(9);
188 for (const pull of pulls.slice(0, 9)) expect(merged(repo, pull)).toBe(true);
189 expect(merged(repo, pulls[9]!)).toBe(false);
190 expect(latestDecision(app.db, pulls[9]!.id, "land")?.answer).toBe("red");
191 });
192
193 test("/boards shows the gates in flight and the minutes from approval to landing", async () => {
194 const repo = await autonomousRepo("on-boards");
195 const pull = await greenPull(repo, "a.ts");
196 submitReview(app.db, {
197 issueId: pull.id,
198 authorId: alice.id,
199 state: "approved",
200 commitOid: pull.headOid,
201 createdAt: new Date(Date.now() - 12 * 60_000).toISOString(),
202 });
203 const sweep = sweepGreen(app, repo);
204 await letGo(1);
205 expect(await sweep).toBe(1);
206 const row = readQueues(app.db, new Date()).find((r) => r.repo === `alice/${repo.name}`);
207 expect(row).toMatchObject({ queued: 0, gating: 0, gates: 0, landedHour: 1, approvedToLanded: 12 });
208 // The terminal's Flow dialog says the same as the page.
209 const line = flowLines(app.db).find((l) => l.startsWith(`alice/${repo.name} `));
210 expect(line?.split(/\s+/)).toEqual([`alice/${repo.name}`, "0", "0", "0", "1", "12"]);
211 const signin = await fetch(`${servers.url}/signin`, {
212 method: "POST",
213 redirect: "manual",
214 headers: { "content-type": "application/x-www-form-urlencoded", origin: servers.url },
215 body: new URLSearchParams({ username: "alice", password: "a-good-passphrase" }),
216 });
217 const cookie = (signin.headers.get("set-cookie") ?? "").split(";")[0] ?? "";
218 const page = await (await fetch(`${servers.url}/boards`, { headers: { cookie } })).text();
219 expect(page).toMatch(
220 /data-queue="alice\/on-boards">[\s\S]*?data-gates>0<[\s\S]*?data-landed>1<[\s\S]*?data-approved-to-landed>12 min</,
221 );
222 });
223 });
224
225 describe("the queue's extra gates and the pull checks' slots", () => {
226 let dataDir: string;
227 let dir: string;
228 let app: App;
229 let checks: ChecksHandle | undefined;
230 const git = (...args: string[]) =>
231 Bun.spawnSync(["git", "-C", dir, "-c", "user.name=t", "-c", "user.email=t@t", ...args]).stdout.toString().trim();
232
233 beforeAll(async () => {
234 dataDir = await mkdtemp(join(tmpdir(), "forge-merge-queue-slots-"));
235 app = createApp({ dataDir, port: 0 });
236 dir = join(dataDir, "line.git");
237 Bun.spawnSync(["git", "init", "-q", "-b", "main", dir]);
238 git("commit", "-q", "--allow-empty", "-m", "start");
239 });
240
241 afterAll(async () => {
242 await checks?.stop({ drain: false });
243 await app.close?.();
244 await rm(dataDir, { recursive: true, force: true });
245 });
246
247 /** A pull's head: one file added on main. */
248 async function headAdding(name: string): Promise<string> {
249 git("checkout", "-q", "-B", name, "main");
250 await Bun.write(join(dir, `${name}.txt`), `${name}\n`);
251 git("add", `${name}.txt`);
252 git("commit", "-q", "-m", name);
253 git("checkout", "-q", "main");
254 return git("rev-parse", name);
255 }
256
257 /**
258 * One pull joins and gates; a second joins while the gate is held. How
259 * many gates were open at once, and the spare slots each gate saw.
260 */
261 async function gatesAtOnce(concurrency: number): Promise<{ most: number; spare: number[] }> {
262 await checks?.stop({ drain: false });
263 checks = startChecks(app, { forgeUrl: "http://127.0.0.1:9", concurrency });
264 const handle = checks;
265 let open = 0;
266 let most = 0;
267 const spare: number[] = [];
268 let release = (): void => {};
269 const held = new Promise<void>((r) => (release = r));
270 const gate = async () => {
271 open++;
272 most = Math.max(most, open);
273 spare.push(handle.spareSlots());
274 await held;
275 open--;
276 return { lands: false, interrupted: true } as const;
277 };
278 const entry = async (n: number) => ({
279 pull: { id: `slots-${concurrency}-${n}`, number: n, baseBranch: "main", headOid: await headAdding(`c${concurrency}-${n}`) } as Pull,
280 dir,
281 land: async () => undefined,
282 release: () => {},
283 handBack: () => {},
284 });
285 const options = { depth: 2, room: () => handle.spareSlots() };
286 const repoId = `slots-${concurrency}`;
287 const [first, second] = [await entry(1), await entry(2)];
288 const one = enqueue(repoId, [first], gate, options);
289 for (let i = 0; i < 2000 && open < 1; i++) await Bun.sleep(5);
290 const two = enqueue(repoId, [second], gate, options);
291 // Either it gates beside the first, or it still waits in line for it.
292 for (let i = 0; i < 100 && open < 2; i++) await Bun.sleep(5);
293 if (open < 2) expect(queueStatus().find((s) => s.repoId === repoId)?.queued).toBe(1);
294 release();
295 expect([...(await one), ...(await two)]).toEqual([false, false]);
296 expect(extraGates()).toBe(0);
297 return { most, spare };
298 }
299
300 test("a spare slot lets the next candidate gate beside the first, and it holds that slot", async () => {
301 const { most, spare } = await gatesAtOnce(7);
302 expect(most).toBe(2);
303 // The second gate holds a slot: one fewer is spare while both run.
304 // (The first is cut off, so the second is formed again on main after.)
305 expect(spare.slice(0, 2)).toEqual([7 - PULL_RESERVE, 7 - PULL_RESERVE - 1]);
306 });
307
308 test("two slots are kept for pull checks: with two, the queue gates one candidate at a time", async () => {
309 const { most, spare } = await gatesAtOnce(2);
310 expect(most).toBe(1);
311 expect(spare.every((n) => n === 0)).toBe(true);
312 });
313 });
test/merge-queue.test.ts +47 −4
@@ -206,6 +206,49 @@ describe("the merge queue", () => {
206206 expect(queue.some((l) => l.includes("[queue] gated 2 pulls: green"))).toBe(true);
207207 });
208208
209 test("after a red batch, the good pulls go back into ONE candidate up to the cap, not one gate each (#4053)", async () => {
210 const repo = await autonomousRepo("rebatch");
211 const pulls = [];
212 for (let i = 0; i < 12; i++) pulls.push(await greenPull(repo, i === 1 ? "red-b.ts" : `q${i}.ts`));
213 const said: string[] = [];
214 const info = console.info;
215 console.info = (line: string) => said.push(line);
216 try {
217 expect(await sweepGreen(app, repo)).toBe(11);
218 } finally {
219 console.info = info;
220 }
221 const sizes = said.flatMap((l) => /\[queue\] gating (\d+) pulls? since/.exec(l)?.[1] ?? []).map(Number);
222 // The eight are red; halving finds the culprit (4, 2, 1); then the six
223 // left of the eight and two of the four behind are one candidate.
224 expect(sizes).toEqual([8, 4, 2, 1, 8, 2]);
225 for (const [i, pull] of pulls.entries()) expect(merged(repo, pull)).toBe(i !== 1);
226 expect(latestDecision(app.db, pulls[1]!.id, "land")?.answer).toBe("red");
227 });
228
229 test("the sweep reads every open pull: an approved one older than the newest fifty is queued with the rest (#4053)", async () => {
230 const repo = await autonomousRepo("many-open");
231 const old = await greenPull(repo, "old.ts");
232 // Fifty newer pulls, none of them checked: the sweep passes them by.
233 for (let i = 0; i < 50; i++) {
234 const pushed = await commitToBranch(app, alice, repo, {
235 branch: `feature/n${i}`,
236 base: "main",
237 message: `n${i}`,
238 files: [{ path: `n${i}.ts`, content: `export const n = ${i};\n` }],
239 });
240 if (!pushed.ok) throw new Error(JSON.stringify(pushed.error));
241 const opened = await openPull(app, alice, repo, { title: `n${i}`, baseBranch: "main", headBranch: `feature/n${i}` });
242 if (!opened.ok) throw new Error(String(opened.error));
243 }
244 const fresh = await greenPull(repo, "fresh.ts");
245 expect(await sweepGreen(app, repo)).toBe(2);
246 expect(merged(repo, old)).toBe(true);
247 expect(merged(repo, fresh)).toBe(true);
248 // One candidate carried both.
249 expect(gatesOf(repo)).toHaveLength(1);
250 });
251
209252 test("one red among four: three land, the culprit is handed back, three gate runs", async () => {
210253 const repo = await autonomousRepo("one-red");
211254 const pulls = [];
@@ -347,8 +390,8 @@ describe("the merge queue", () => {
347390 /data-queue="alice\/on-flow">[\s\S]*?data-queued>2<[\s\S]*?data-gating>1<[\s\S]*?data-landed>0</,
348391 );
349392 // The terminal's Flow dialog and the MCP tool say the same.
350 expect(terminalLine(repo)?.split(/\s+/)).toEqual([`alice/${repo.name}`, "2", "1", "0"]);
351 expect(await mcpQueue()).toContainEqual({ repo: `alice/${repo.name}`, queued: 2, gating: 1, landedHour: 0 });
393 expect(terminalLine(repo)?.split(/\s+/)).toEqual([`alice/${repo.name}`, "2", "1", "1", "0", "—"]);
394 expect(await mcpQueue()).toContainEqual({ repo: `alice/${repo.name}`, queued: 2, gating: 1, gates: 1, landedHour: 0, approvedToLanded: null });
352395 held = undefined;
353396 release();
354397 expect(await landing).toBe(true);
@@ -360,8 +403,8 @@ describe("the merge queue", () => {
360403 gating: 0,
361404 landedHour: 3,
362405 });
363 expect(terminalLine(repo)?.split(/\s+/)).toEqual([`alice/${repo.name}`, "0", "0", "3"]);
364 expect(await mcpQueue()).toContainEqual({ repo: `alice/${repo.name}`, queued: 0, gating: 0, landedHour: 3 });
406 expect(terminalLine(repo)?.split(/\s+/)).toEqual([`alice/${repo.name}`, "0", "0", "0", "3", "—"]);
407 expect(await mcpQueue()).toContainEqual({ repo: `alice/${repo.name}`, queued: 0, gating: 0, gates: 0, landedHour: 3, approvedToLanded: null });
365408 });
366409
367410 /** Ages the pull's newest land decision past the re-gate wait. */

↑↓ navigate ↵ open ⌘↵ its app esc close

Keyboard shortcuts

Docs