trevato/forge

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

More
Forks1g fMapg aRuns21g rEventsFlows

The merge queue survives a rollout: gates in flight resume on the new pod instead of starting over #6616

Merged masoncrew opened this in answer to trevatoowner

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

ConversationFiles
Merged

mason/forge:mason/6611-the-merge-queue-survives-a-rollout-gates → main 5 commits fork 1 thread · 1 unresolved

  • Merged. mason/6611-the-merge-queue-survives-a-rollout-gates was merged into main as 1ddfd28.

Review

Sign in to review this change.
12 files changed +1297 −89
docs/ARCHITECTURE.md +4 −2
@@ -416,8 +416,10 @@ ends every open event stream with a resume hint (`retry: 1000`, then
416416 process — from the last id it saw; the crew stops taking runs and lets the
417417 ones it holds finish and land (`crew.stop({ drain: true })`, bounded); a
418418 check in flight is cut and handed back to the queue for the new process.
419 A gate on land in flight is cut too, and is no verdict: no red is told, and
420 the new process's merge queue gates it again (#3155). Its app processes stop
419 A merge-queue gate in flight runs on through the same drain; its verdict is
420 kept in its `queue_gate` row for the new process to act on (#6611). One the
421 drain cuts off is no verdict: no red is told, and the new process's merge
422 queue gates the same candidate again (#3155). Its app processes stop
421423 (the new process already runs its own), and SSH sessions are kept until
422424 they leave. However little it holds, its listener stays open
423425 for `FORGE_DRAIN_LINGER_S` (10 s) after the SIGTERM: routers hear that it
docs/CREW.md +16 −1
@@ -203,7 +203,22 @@ candidate`, or `red on its own merge`), and the tick line counts them as
203203 alone, only once no fresh pull waits, and nothing is stacked on it, so a red
204204 re-check never cuts off a batch of fresh pulls; a push forgets it. A gate cut off by the forge stopping (a
205205 roll, a drain) or by the cluster losing its Job is no red at all: the next
206 sweep gates it again. So is a gate that has not answered in 15 minutes
206 sweep gates it again. A roll does not cost a gate round (#6611): each gate's
207 candidate — its pulls and heads, the base, each step's commit and tree, the
208 lane's run id (`gate-<tip>`) and when it started — is a `queue_gate` row, and
209 the red marks are `queue_red` rows by pull, head and base
210 (`src/checks/queue_store.ts`). A stopping forge's queue acts on nothing more
211 (`stopQueue`) but its gates run on through the checks' drain, and a verdict
212 that comes in is written to the row. The new process resumes the recorded
213 candidate when its pulls are back at the same heads and main has not moved:
214 the same commits, not a candidate planned again. It acts on a kept verdict,
215 adopts the run when the lane still has it (`[queue] resuming N pulls`, and
216 `adopted=gate-…` on the verdict line), or gates the same tip again when the
217 run is gone. A record whose base moved is forgotten. A pull waiting in the line that is
218 held, closed, pushed to, no longer green on its head or sent back for
219 changes is taken out on the next tick (`pruneLine`, `[queue] taken out of
220 the line`), before the next batch is formed, and the tick line's waiting
221 count drops with it; a pull already in a candidate is left to its gate. So is a gate that has not answered in 15 minutes
207222 (`GATE_DEADLINE_MS`): it is abandoned and its check aborted, the log says
208223 `[queue] candidate abandoned`, and the next sweep forms the line again. Pulls
209224 that turn green while a candidate is gating wait for the next one. Each
src/checks/landing.ts +44 −2
@@ -28,7 +28,9 @@ import { failuresOf } from "./failures.ts";
2828 import { keepForgeGate } from "./gate_last.ts";
2929 import { gateReason } from "./gate_reason.ts";
3030 import { keepGateLog } from "./gate_log.ts";
31 import { checksFor } from "./service.ts";
31 import { GATE_DEADLINE_MS, type GateRecord } from "./queue.ts";
32 import { gateRecorded, gateRunning, recordedVerdict } from "./queue_store.ts";
33 import { checksFor, type GateDone } from "./service.ts";
3234 import { landVerdict, type LandVerdict } from "./tiers.ts";
3335 import { ownProbeHeaders } from "../kernel/probe.ts";
3436
@@ -56,7 +58,47 @@ export async function gateCandidate(
5658 const declared = parseCheck(forgeJson, packageJson);
5759 if (!declared.ok || declared.value === undefined || declared.value.fast === null)
5860 return { lands: true };
59 const done = await checks.gate(repo, pull, commitOid, signal);
61 return gateVerdict(app, repo, commitOid, await checks.gate(repo, pull, commitOid, signal));
62 }
63
64 /** How long the queue waits between reads of a gate another forge runs (#6611). */
65 export const ADOPT_POLL_MS = 5_000;
66
67 /**
68 * A recorded merge-queue gate the check lane still runs (#6611), adopted:
69 * this forge's own run of its tip, waited on; or the forge before this one
70 * still draining it, whose verdict comes into the record. Undefined when
71 * the run is gone and the queue must gate the candidate again.
72 */
73 export async function adoptGate(
74 app: App,
75 repo: Repo,
76 record: GateRecord,
77 signal: AbortSignal,
78 pollMs = ADOPT_POLL_MS,
79 ): Promise<LandVerdict | undefined> {
80 const here = checksFor(app)?.adopt?.(record.tipOid);
81 if (here !== undefined) {
82 const done = await here;
83 return done.interrupted === true ? undefined : gateVerdict(app, repo, record.tipOid, done);
84 }
85 // The trace closes a moment before the verdict is kept: read once more after it.
86 let open = gateRunning(app.db, record.tipOid, GATE_DEADLINE_MS);
87 if (!open) return undefined;
88 while (!signal.aborted) {
89 await Bun.sleep(pollMs);
90 // The record gone: the forge that ran it acted on its verdict itself
91 // (its queue was not stopped). No verdict here, and nothing to gate again.
92 if (!gateRecorded(app.db, record.id)) return { lands: false, interrupted: true };
93 const verdict = recordedVerdict(app.db, record.id);
94 if (verdict !== undefined || !open) return verdict;
95 open = gateRunning(app.db, record.tipOid, GATE_DEADLINE_MS);
96 }
97 return undefined;
98 }
99
100 /** What the lane's answer on `commitOid` decides: its log kept, its red named. */
101 function gateVerdict(app: App, repo: Repo, commitOid: string, done: GateDone): LandVerdict {
60102 // Cut off, not judged: no verdict, and not the gate log a red reads (#3155).
61103 if (done.interrupted === true) return { lands: false, interrupted: true };
62104 keepForgeGate(app.config.dataDir, app.config.forgeRepo, `${repo.ownerName}/${repo.name}`, done.lines);
src/checks/queue.ts +295 −28
@@ -58,6 +58,7 @@
5858 import type { Pull } from "../core/pulls.ts";
5959 import { runGit } from "../git/env.ts";
6060 import { mergeable, resolveOid, treeOf } from "../git/merge.ts";
61 import { id } from "../kernel/id.ts";
6162 import { createLog } from "../kernel/log.ts";
6263 import { conflictWords, reasonWords, searched } from "./gate_reason.ts";
6364 import type { LandVerdict } from "./tiers.ts";
@@ -157,6 +158,65 @@ export interface QueueOptions {
157158 * kept beyond this process (#6427).
158159 */
159160 readonly onBatch?: (repoId: string, red: boolean) => void;
161 /** Where the line's gates and red marks outlive this process (#6611). */
162 readonly store?: QueueStore;
163 /**
164 * The verdict of a recorded gate still running on the check lane — in
165 * this process or the one before it — or undefined when its run is gone
166 * and the candidate must be gated again (#6611).
167 */
168 readonly adopt?: (record: GateRecord, signal: AbortSignal) => Promise<LandVerdict | undefined>;
169 }
170
171 /** A step as recorded: which pull at which head, and the commit it made. */
172 export interface RecordedStep {
173 readonly pullId: string;
174 readonly headOid: string;
175 readonly baseOid: string;
176 readonly treeOid: string;
177 readonly commitOid: string;
178 }
179
180 /**
181 * A gate as kept beyond this process (#6611): its candidate exactly as
182 * built, the run the check lane gates it as, and its verdict once it has
183 * one. A forge that carries on rebuilds the candidate from it, not the line.
184 */
185 export interface GateRecord {
186 readonly id: string;
187 readonly repoId: string;
188 /** The check lane's name for the gate: `gate-` and the tip's first 12. */
189 readonly runId: string;
190 readonly baseOid: string;
191 readonly baseTreeOid: string;
192 readonly tipOid: string;
193 /** The tree the gate judges: on a speculative candidate, a tree main never had. */
194 readonly treeOid: string;
195 readonly steps: readonly RecordedStep[];
196 readonly startedAt: string;
197 /** Green or red, once the gate answered; never a cut-off. */
198 readonly verdict?: Exclude<LandVerdict, { interrupted: true }>;
199 }
200
201 /** What the line keeps outside memory (`checks/queue_store.ts`). */
202 export interface QueueStore {
203 /** A gate started. */
204 began(record: GateRecord): void;
205 /** A gate answered after its line stopped: kept for the line that carries on. */
206 settled(id: string, verdict: Exclude<LandVerdict, { interrupted: true }>): void;
207 /** A gate acted on, abandoned, or no longer anyone's to resume. */
208 forget(id: string): void;
209 /** The repository's recorded gates, oldest first. */
210 gates(repoId: string): GateRecord[];
211 /** Mark a pull red at a head and base, replacing its mark; undefined clears it. */
212 red(repoId: string, pullId: string, mark: Remembered | undefined): void;
213 /** The repository's red marks, by pull id. */
214 reds(repoId: string): Map<string, Remembered>;
215 }
216
217 /** The check lane's run id for a gate of `tipOid` (`checks/service.ts` names it so). */
218 export function gateRunId(tipOid: string): string {
219 return `gate-${tipOid.slice(0, 12)}`;
160220 }
161221
162222 /**
@@ -173,8 +233,12 @@ export function gateDeadlineForTest(ms: number, arm?: Schedule): () => void {
173233 };
174234 }
175235
176 /** The gate as the line asks it: the deadline is the line's own. */
177 type Gated = (candidate: Candidate, signal?: AbortSignal) => Promise<LandVerdict>;
236 /**
237 * The gate as the line asks it: the deadline is the line's own. `resumed`
238 * is the record of a gate an earlier line started on this candidate: its
239 * verdict, or its run adopted, before a gate of its own.
240 */
241 type Gated = (candidate: Candidate, signal?: AbortSignal, resumed?: GateRecord) => Promise<LandVerdict>;
178242
179243 /** Check slots free now for one more gate beyond a line's first. */
180244 type Room = () => number;
@@ -187,7 +251,7 @@ interface Waiting {
187251 }
188252
189253 /** A pull the line handed back red, while its head is the same (#4235). */
190 interface Remembered {
254 export interface Remembered {
191255 readonly head: string;
192256 /** The base tip it went red on: a moved base is a new try (#6493). */
193257 readonly base: string;
@@ -238,6 +302,12 @@ interface Line {
238302 draining: boolean;
239303 /** Wakes a drain waiting on a gate when pulls join the line. */
240304 wake: (() => void) | undefined;
305 /** Where its gates and red marks are kept, when anywhere (#6611). */
306 store: QueueStore | undefined;
307 /** Gates an earlier line recorded and this one may resume, oldest first. */
308 kept: GateRecord[];
309 /** Stopped (`stopQueue`): its gates run on, but it acts on no verdict. */
310 stopped: boolean;
241311 }
242312
243313 /** A candidate at the gate in the pipeline, and the pulls it took from the line. */
@@ -315,9 +385,14 @@ const IDENTITY = {
315385
316386 const UNDECIDED: LandVerdict = { lands: false, interrupted: true };
317387
318 function lineOf(repoId: string): Line {
388 /**
389 * The repository's line. A new one starts from what the store kept (#6611):
390 * the red marks, and the gates an earlier line left to resume.
391 */
392 function lineOf(repoId: string, store: QueueStore | undefined): Line {
319393 let line = lines.get(repoId);
320394 if (line === undefined) {
395 const reds = kept(() => store?.reds(repoId)) ?? new Map<string, Remembered>();
321396 line = {
322397 repoId,
323398 waiting: [],
@@ -325,17 +400,61 @@ function lineOf(repoId: string): Line {
325400 flights: new Set(),
326401 forming: 0,
327402 gates: 0,
328 reds: new Map(),
403 reds,
329404 suspects: new Map(),
330405 batches: [],
331406 draining: false,
332407 wake: undefined,
408 store,
409 kept: kept(() => store?.gates(repoId)) ?? [],
410 stopped: false,
333411 };
334412 lines.set(repoId, line);
335413 }
414 line.store ??= store;
336415 return line;
337416 }
338417
418 /** A store call: what it answers, or undefined, logged, when it fails. The line goes on in memory. */
419 function kept<T>(fn: () => T): T | undefined {
420 try {
421 return fn();
422 } catch (error) {
423 log.warn("merge queue state not kept", { error: String(error) });
424 return undefined;
425 }
426 }
427
428 /** Mark a pull red at its head and base, here and in the store. */
429 function markRed(line: Line, pullId: string, mark: Remembered): void {
430 line.reds.set(pullId, mark);
431 kept(() => line.store?.red(line.repoId, pullId, mark));
432 }
433
434 /** Forget a pull's red mark, here and in the store. */
435 function clearRed(line: Line, pullId: string): void {
436 if (!line.reds.delete(pullId)) return;
437 kept(() => line.store?.red(line.repoId, pullId, undefined));
438 }
439
440 /**
441 * Stop every line (#6611), as the forge stopping does first: what waits is
442 * answered `false`, and no verdict from here on is acted on — the gates in
443 * flight run on, and one that answers is kept in the store for the forge
444 * that carries on. Lines asked for after this start afresh from the store.
445 */
446 export function stopQueue(): void {
447 for (const line of lines.values()) {
448 line.stopped = true;
449 const waiting = line.waiting;
450 line.waiting = [];
451 for (const w of waiting) w.settle(false);
452 line.wake?.();
453 line.wake = undefined;
454 }
455 lines.clear();
456 }
457
339458 /**
340459 * Join the repository's line, in this order. Each answer is whether that
341460 * pull landed. A pull already in the line is answered `false` at once: the
@@ -347,7 +466,7 @@ export async function enqueue(
347466 gate: GateCandidate,
348467 options: QueueOptions = {},
349468 ): Promise<boolean[]> {
350 const line = lineOf(repoId);
469 const line = lineOf(repoId, options.store);
351470 // A red mark names the base it went red on; the tip is read only when a
352471 // mark is there to compare it with.
353472 const tips = new Map<string, string | undefined>();
@@ -361,7 +480,7 @@ export async function enqueue(
361480 const red = line.reds.get(entry.pull.id);
362481 const tip = tips.get(entry.pull.baseBranch + "\0" + entry.dir);
363482 const held = red !== undefined && red.head === entry.pull.headOid && (tip === undefined || red.base === tip);
364 if (red !== undefined && !held) line.reds.delete(entry.pull.id);
483 if (red !== undefined && !held) clearRed(line, entry.pull.id);
365484 const recheck = held;
366485 if (line.suspects.get(entry.pull.id)?.head !== entry.pull.headOid) line.suspects.delete(entry.pull.id);
367486 return new Promise<boolean>((resolve) =>
@@ -371,7 +490,7 @@ export async function enqueue(
371490 settle: (landed) => {
372491 line.holds.delete(entry.pull.id);
373492 if (landed) {
374 line.reds.delete(entry.pull.id);
493 clearRed(line, entry.pull.id);
375494 line.suspects.delete(entry.pull.id);
376495 }
377496 resolve(landed);
@@ -387,6 +506,32 @@ export async function enqueue(
387506 return Promise.all(answers);
388507 }
389508
509 /** The pulls waiting in a repository's line, not yet in a candidate: as they joined. */
510 export function waitingPulls(repoId: string): Pull[] {
511 return (lines.get(repoId)?.waiting ?? []).map((w) => w.entry.pull);
512 }
513
514 /**
515 * Take pulls out of the line before they reach a gate (#6611): held,
516 * closed, pushed to or no longer approved, a pull that waits on would be
517 * gated in the next batch all the same. Each is answered `false`; a pull
518 * already in a candidate is left to its gate. How many were taken out.
519 */
520 export function dropFromLine(repoId: string, pullIds: ReadonlySet<string>): number {
521 const line = lines.get(repoId);
522 if (line === undefined) return 0;
523 const gone = line.waiting.filter((w) => pullIds.has(w.entry.pull.id));
524 if (gone.length === 0) return 0;
525 line.waiting = line.waiting.filter((w) => !pullIds.has(w.entry.pull.id));
526 for (const w of gone) w.settle(false);
527 log.info("taken out of the line", {
528 repo: repoId,
529 pulls: gone.map((w) => `#${w.entry.pull.number}`).join(","),
530 waiting: line.waiting.length,
531 });
532 return gone.length;
533 }
534
390535 /** What the flow panel shows of each repository's line. */
391536 export interface QueueStatus {
392537 readonly repoId: string;
@@ -487,10 +632,10 @@ export function gateRuns(repoId: string): number {
487632 * setup, a timeout. Red that way twice is infrastructure — no verdict, as a
488633 * gate cut off is not: no pull is handed back and nothing is searched.
489634 */
490 function counted(line: Line, gate: GateCandidate): Gated {
491 const once = attempt(line, gate);
492 return async (candidate, signal) => {
493 const first = await once(candidate, signal);
635 function counted(line: Line, gate: GateCandidate, adopt: QueueOptions["adopt"]): Gated {
636 const once = attempt(line, gate, adopt);
637 return async (candidate, signal, resumed) => {
638 const first = await once(candidate, signal, resumed);
494639 if (!machines(first) || signal?.aborted === true) return first;
495640 const again = await once(candidate, signal);
496641 if (!machines(again)) return again;
@@ -507,20 +652,31 @@ function counted(line: Line, gate: GateCandidate): Gated {
507652 * minutes, #3430), and abandoned at the deadline or when `signal` aborts —
508653 * a speculative candidate discarded. Abandoned is no verdict.
509654 */
510 function attempt(line: Line, gate: GateCandidate): Gated {
511 return async (candidate, signal) => {
655 function attempt(line: Line, gate: GateCandidate, adopt: QueueOptions["adopt"]): Gated {
656 return async (candidate, signal, resumed) => {
657 // Answered before this line was: acted on now, nothing gated (#6611).
658 if (resumed?.verdict !== undefined) {
659 log.info(`resumed a gate's ${outcome(resumed.verdict)} verdict`, {
660 repo: line.repoId,
661 pulls: candidate.steps.map((s) => `#${s.entry.pull.number}`).join(","),
662 });
663 kept(() => line.store?.forget(resumed.id));
664 return resumed.verdict;
665 }
512666 const beyond = line.flights.size > 0;
513 const flight: InFlight = { pulls: candidate.steps.map((s) => s.entry.pull.id), since: Date.now() };
667 const since = resumed === undefined ? Date.now() : Date.parse(resumed.startedAt) || Date.now();
668 const flight: InFlight = { pulls: candidate.steps.map((s) => s.entry.pull.id), since };
514669 line.flights.add(flight);
515 line.gates++;
516670 if (beyond) extra++;
517671 const n = candidate.steps.length;
518672 const pulls = candidate.steps.map((s) => `#${s.entry.pull.number}`).join(",");
519 log.info(`gating ${pullsWord(n)} since ${hhmm(flight.since)}`, {
673 log.info(`${resumed === undefined ? "gating" : "resuming"} ${pullsWord(n)} since ${hhmm(flight.since)}`, {
520674 repo: line.repoId,
521675 pulls,
522676 waiting: line.waiting.length,
523677 });
678 const record = resumed ?? recordOf(line.repoId, candidate, flight.since);
679 if (resumed === undefined) kept(() => line.store?.began(record));
524680 const abandon = new AbortController();
525681 let timer: ReturnType<typeof setTimeout> | undefined;
526682 let dropped: (() => void) | undefined;
@@ -545,12 +701,25 @@ function attempt(line: Line, gate: GateCandidate): Gated {
545701 else signal?.addEventListener("abort", dropped, { once: true });
546702 });
547703 try {
548 const verdict = await Promise.race([gate(candidate, abandon.signal), cut]);
704 // A recorded gate still running on the check lane is adopted, not
705 // started again; one whose run is gone is gated again as it was (#6611).
706 const adopted =
707 resumed === undefined || adopt === undefined
708 ? undefined
709 : await Promise.race([adopt(resumed, abandon.signal), cut]);
710 if (adopted === undefined) line.gates++;
711 const verdict = adopted ?? (await Promise.race([gate(candidate, abandon.signal), cut]));
549712 log.info(`gated ${pullsWord(n)}: ${outcome(verdict)}`, {
550713 repo: line.repoId,
551714 pulls,
552715 took: `${Math.round((Date.now() - flight.since) / 1000)}s`,
716 ...(adopted === undefined ? {} : { adopted: record.runId }),
553717 });
718 // A stopped line acts on nothing: a verdict is kept for the line that
719 // carries on, and so is a gate its forge stopping cut off. Anything
720 // else is acted on now, or was abandoned: not to resume.
721 if (line.stopped && !("interrupted" in verdict)) kept(() => line.store?.settled(record.id, verdict));
722 else if (!line.stopped || abandon.signal.aborted) kept(() => line.store?.forget(record.id));
554723 return verdict;
555724 } finally {
556725 clearTimeout(timer);
@@ -562,6 +731,27 @@ function attempt(line: Line, gate: GateCandidate): Gated {
562731 };
563732 }
564733
734 /** The record of a gate starting now on `candidate`. */
735 function recordOf(repoId: string, candidate: Candidate, since: number): GateRecord {
736 return {
737 id: id(),
738 repoId,
739 runId: gateRunId(candidate.tipOid),
740 baseOid: candidate.baseOid,
741 baseTreeOid: candidate.baseTreeOid,
742 tipOid: candidate.tipOid,
743 treeOid: candidate.treeOid,
744 steps: candidate.steps.map((s) => ({
745 pullId: s.entry.pull.id,
746 headOid: s.entry.pull.headOid,
747 baseOid: s.baseOid,
748 treeOid: s.treeOid,
749 commitOid: s.commitOid,
750 })),
751 startedAt: new Date(since).toISOString(),
752 };
753 }
754
565755 /** A drain waiting on its gate, woken by pulls joining the line. */
566756 const ARRIVED = Symbol("arrived");
567757
@@ -578,13 +768,19 @@ const ARRIVED = Symbol("arrived");
578768 */
579769 async function drain(line: Line, gate: GateCandidate, options: QueueOptions): Promise<void> {
580770 line.draining = true;
581 const gated = counted(line, gate);
771 const gated = counted(line, gate, options.adopt);
582772 const depth = Math.max(1, Math.floor(options.depth ?? DEFAULT_DEPTH));
583773 const room: Room = options.room ?? (() => 0);
584774 const ahead: Flight[] = [];
585775 const solos = new Set<Promise<void>>();
586776 try {
587777 while (line.waiting.length > 0 || ahead.length > 0 || solos.size > 0) {
778 // Stopped: what is at the gate is answered `false` and left running,
779 // for the forge that carries on to adopt (#6611).
780 if (line.stopped) {
781 for (const f of ahead.splice(0)) for (const w of f.steps) w.settle(false);
782 return;
783 }
588784 for (const w of line.waiting.filter((w) => suspectOf(line, w)?.alone === true)) {
589785 if (line.flights.size > 0 && room() <= 0) break;
590786 line.waiting = line.waiting.filter((x) => x !== w);
@@ -607,19 +803,17 @@ async function drain(line: Line, gate: GateCandidate, options: QueueOptions): Pr
607803 if (flight !== undefined) ahead.push(flight);
608804 }
609805 const head = ahead[0];
610 const arrived =
611 head === undefined || ahead.length < depth || solos.size > 0
612 ? new Promise<typeof ARRIVED>((resolve) => (line.wake = () => resolve(ARRIVED)))
613 : undefined;
806 // Pulls joining, a suspect coming back or the line stopping wake it.
807 const arrived = new Promise<typeof ARRIVED>((resolve) => (line.wake = () => resolve(ARRIVED)));
614808 if (head === undefined) {
615809 // Only suspects at the gate, or waiting on a slot: wait for one to come back.
616810 if (solos.size > 0) await arrived;
617811 line.wake = undefined;
618812 continue;
619813 }
620 const verdict = arrived === undefined ? await head.verdict : await Promise.race([head.verdict, arrived]);
814 const verdict = await Promise.race([head.verdict, arrived]);
621815 line.wake = undefined;
622 if (verdict === ARRIVED) continue;
816 if (verdict === ARRIVED || line.stopped) continue;
623817 ahead.shift();
624818 let left: Waiting[];
625819 try {
@@ -671,6 +865,7 @@ async function solo(line: Line, gated: Gated, w: Waiting): Promise<{ readonly do
671865 const done = gated(on)
672866 .then((verdict) => {
673867 const pull = w.entry.pull;
868 if (line.stopped) return w.settle(false);
674869 if (verdict.lands) {
675870 line.suspects.set(pull.id, { head: pull.headOid, reds: 0, alone: false, cleared: true });
676871 line.waiting.unshift(w);
@@ -681,7 +876,7 @@ async function solo(line: Line, gated: Gated, w: Waiting): Promise<{ readonly do
681876 return;
682877 }
683878 line.suspects.delete(pull.id);
684 line.reds.set(pull.id, { head: pull.headOid, base: on.baseOid, together: [] });
879 markRed(line, pull.id, { head: pull.headOid, base: on.baseOid, together: [] });
685880 log.info("red alone", { repo: line.repoId, pull: pull.number });
686881 w.entry.handBack(verdict, on.baseOid, []);
687882 w.settle(false);
@@ -725,6 +920,26 @@ function explained(line: Line, steps: readonly Waiting[]): void {
725920 * undefined when nothing of the batch could be built.
726921 */
727922 async function launch(line: Line, gated: Gated, on: Flight | undefined): Promise<Flight | undefined | "apart"> {
923 if (on === undefined && line.kept.length > 0) {
924 line.forming++;
925 try {
926 const resumed = await resumable(line);
927 if (resumed !== undefined) {
928 const { record, batch, candidate } = resumed;
929 line.waiting = line.waiting.filter((w) => !batch.includes(w));
930 const abort = new AbortController();
931 const verdict = gated(candidate, abort.signal, record).catch((error: unknown) => {
932 log.warn("merge queue round failed", { error: String(error) });
933 return UNDECIDED;
934 });
935 const recheck = batch.length === 1 && batch[0]!.recheck;
936 return { steps: batch, candidate, speculative: false, recheck, branch: batch[0]!.entry.pull.baseBranch, verdict, abort };
937 }
938 } finally {
939 line.forming--;
940 gateFreed();
941 }
942 }
728943 const cleared = (w: Waiting): boolean => suspectOf(line, w)?.cleared === true;
729944 const open = line.waiting.filter((w) => suspectOf(line, w)?.alone !== true);
730945 const front = open.find((w) => !w.recheck && !cleared(w)) ?? open.find((w) => !w.recheck) ?? open[0]!;
@@ -763,6 +978,54 @@ async function launch(line: Line, gated: Gated, on: Flight | undefined): Promise
763978 return { steps, candidate, speculative: on !== undefined, recheck: front.recheck, branch, verdict, abort };
764979 }
765980
981 /**
982 * The first gate an earlier line recorded that this one can resume (#6611):
983 * every pull of it waiting at the head it was gated at, none a suspect to
984 * gate alone, and the base still at the tip it was built on. Its candidate
985 * is rebuilt from the record — the very commits that were gated — so it is
986 * not planned again. A record that cannot be resumed is forgotten.
987 */
988 async function resumable(
989 line: Line,
990 ): Promise<{ record: GateRecord; batch: Waiting[]; candidate: Candidate } | undefined> {
991 while (line.kept.length > 0) {
992 const record = line.kept.shift()!;
993 const batch: Waiting[] = [];
994 for (const s of record.steps) {
995 const w = line.waiting.find((w) => w.entry.pull.id === s.pullId && w.entry.pull.headOid === s.headOid);
996 if (w === undefined || suspectOf(line, w)?.alone === true) break;
997 batch.push(w);
998 }
999 const first = batch[0]?.entry;
1000 const whole =
1001 first !== undefined &&
1002 batch.length === record.steps.length &&
1003 batch.every((w) => w.entry.dir === first.dir && w.entry.pull.baseBranch === first.pull.baseBranch) &&
1004 (await resolveOid(first.dir, `refs/heads/${first.pull.baseBranch}`)) === record.baseOid &&
1005 (await resolveOid(first.dir, record.tipOid)) === record.tipOid;
1006 if (!whole) {
1007 kept(() => line.store?.forget(record.id));
1008 continue;
1009 }
1010 const steps = record.steps.map((s, i) => ({
1011 entry: batch[i]!.entry,
1012 baseOid: s.baseOid,
1013 treeOid: s.treeOid,
1014 commitOid: s.commitOid,
1015 }));
1016 const candidate: Candidate = {
1017 dir: first.dir,
1018 baseOid: record.baseOid,
1019 baseTreeOid: record.baseTreeOid,
1020 steps,
1021 tipOid: record.tipOid,
1022 treeOid: record.treeOid,
1023 };
1024 return { record, batch, candidate };
1025 }
1026 return undefined;
1027 }
1028
7661029 /**
7671030 * The pulls of `batch` the candidate carries. The rest — a conflict with
7681031 * the base or an earlier step, or nothing left to land once the steps
@@ -887,6 +1150,10 @@ async function round(
8871150 const steps = inCandidate(batch, candidate);
8881151 const verdict: LandVerdict =
8891152 known.treeOid === candidate.treeOid ? known.red : await gated(candidate);
1153 if (search.line.stopped) {
1154 for (const w of steps) w.settle(false);
1155 return [];
1156 }
8901157 if (verdict.lands) {
8911158 await land(candidate, steps);
8921159 return [];
@@ -926,7 +1193,7 @@ async function judged(
9261193 // with them, or red on its own — not gated alone on the base before
9271194 // them to tell which. Remembered with them either way (#4235).
9281195 const together = landed.map((q) => q.pull.number);
929 search.line.reds.set(culprit.pull.id, { head: culprit.pull.headOid, base: candidate.baseOid, together });
1196 markRed(search.line, culprit.pull.id, { head: culprit.pull.headOid, base: candidate.baseOid, together });
9301197 search.line.suspects.delete(culprit.pull.id);
9311198 log.info(together.length === 0 ? "red on its own merge" : "red on top of its candidate", {
9321199 repo: search.line.repoId,
@@ -946,7 +1213,7 @@ async function judged(
9461213 for (const run of runs) ends.push((ends.at(-1) ?? 0) + run.length);
9471214 const prefixes = ends.slice(0, -1).map((end) => prefixOf(candidate, end));
9481215 const verdicts = await Promise.all(prefixes.map((c) => gated(c)));
949 if (verdicts.some((v) => !v.lands && "interrupted" in v)) {
1216 if (search.line.stopped || verdicts.some((v) => !v.lands && "interrupted" in v)) {
9501217 for (const w of steps) w.settle(false);
9511218 return [];
9521219 }
src/checks/queue_store.ts added +215
@@ -0,0 +1,215 @@
1 /**
2 * The merge queue's state outside memory (#6611): each gate's candidate as
3 * it was built (`queue_gate`), and the line's red marks by pull, head and
4 * base (`queue_red`). A roll used to cost a gate round — the new process
5 * planned the line from nothing and gated it again; with these it resumes
6 * the candidate in flight, adopting its run or its verdict (`checks/queue.ts`).
7 *
8 * Plain rows in, plain records out; what is read back is checked here, the
9 * boundary, so a row this build cannot read is no record at all.
10 */
11
12 import type { Database } from "bun:sqlite";
13 import type { GateRecord, QueueStore, RecordedStep, Remembered } from "./queue.ts";
14 import type { GateReason } from "./gate_reason.ts";
15 import type { LandVerdict } from "./tiers.ts";
16
17 /** How long a gate's record is kept unresumed: past it, its line has long moved on. */
18 export const KEEP_GATE_MS = 24 * 3_600_000;
19
20 type Judged = Exclude<LandVerdict, { interrupted: true }>;
21
22 const OID = /^[0-9a-f]{40}([0-9a-f]{24})?$/;
23
24 interface GateRow {
25 id: string;
26 repo_id: string;
27 run_id: string;
28 base_oid: string;
29 base_tree_oid: string;
30 tip_oid: string;
31 tree_oid: string;
32 steps: string;
33 started_at: string;
34 verdict: string | null;
35 }
36
37 /** The queue's store over this database. */
38 export function queueStore(db: Database, now: () => Date = () => new Date()): QueueStore {
39 return {
40 began: (record) => beginGateRecord(db, record, now()),
41 settled: (id, verdict) => settleGateRecord(db, id, verdict),
42 forget: (id) => void db.query("DELETE FROM queue_gate WHERE id = ?1").run(id),
43 gates: (repoId) => gateRecords(db, repoId),
44 red: (repoId, pullId, mark) => keepRed(db, repoId, pullId, mark, now()),
45 reds: (repoId) => redMarks(db, repoId),
46 };
47 }
48
49 /** Record a gate as it starts; records past `KEEP_GATE_MS` go. */
50 export function beginGateRecord(db: Database, record: GateRecord, now = new Date()): void {
51 db.query("DELETE FROM queue_gate WHERE started_at < ?1").run(new Date(now.getTime() - KEEP_GATE_MS).toISOString());
52 db.query(
53 `INSERT OR REPLACE INTO queue_gate
54 (id, repo_id, run_id, base_oid, base_tree_oid, tip_oid, tree_oid, steps, started_at, verdict)
55 VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, NULL)`,
56 ).run(
57 record.id,
58 record.repoId,
59 record.runId,
60 record.baseOid,
61 record.baseTreeOid,
62 record.tipOid,
63 record.treeOid,
64 JSON.stringify(record.steps),
65 record.startedAt,
66 );
67 }
68
69 /** A gate answered with no line to act on it: its verdict waits in the record. */
70 export function settleGateRecord(db: Database, id: string, verdict: Judged): void {
71 db.query("UPDATE queue_gate SET verdict = ?2 WHERE id = ?1").run(id, JSON.stringify(verdict));
72 }
73
74 /** The repository's recorded gates, oldest first; a row that does not read is left out. */
75 export function gateRecords(db: Database, repoId: string): GateRecord[] {
76 const rows = db
77 .query<GateRow, [string]>("SELECT * FROM queue_gate WHERE repo_id = ?1 ORDER BY started_at, id")
78 .all(repoId);
79 const out: GateRecord[] = [];
80 for (const row of rows) {
81 const record = gateOf(row);
82 if (record !== undefined) out.push(record);
83 }
84 return out;
85 }
86
87 function gateOf(row: GateRow): GateRecord | undefined {
88 const oids = [row.base_oid, row.base_tree_oid, row.tip_oid, row.tree_oid];
89 if (!oids.every((o) => OID.test(o))) return undefined;
90 const steps = stepsOf(row.steps);
91 if (steps === undefined || steps.length === 0) return undefined;
92 const verdict = row.verdict === null ? undefined : verdictOf(row.verdict);
93 if (row.verdict !== null && verdict === undefined) return undefined;
94 return {
95 id: row.id,
96 repoId: row.repo_id,
97 runId: row.run_id,
98 baseOid: row.base_oid,
99 baseTreeOid: row.base_tree_oid,
100 tipOid: row.tip_oid,
101 treeOid: row.tree_oid,
102 steps,
103 startedAt: row.started_at,
104 ...(verdict === undefined ? {} : { verdict }),
105 };
106 }
107
108 function parsed(text: string): unknown {
109 try {
110 return JSON.parse(text);
111 } catch {
112 return undefined;
113 }
114 }
115
116 function stepsOf(text: string): RecordedStep[] | undefined {
117 const list = parsed(text);
118 if (!Array.isArray(list)) return undefined;
119 const steps: RecordedStep[] = [];
120 for (const item of list as unknown[]) {
121 if (typeof item !== "object" || item === null) return undefined;
122 const { pullId, headOid, baseOid, treeOid, commitOid } = item as Record<string, unknown>;
123 if (typeof pullId !== "string" || pullId === "") return undefined;
124 const oids = [headOid, baseOid, treeOid, commitOid];
125 if (!oids.every((o) => typeof o === "string" && OID.test(o))) return undefined;
126 steps.push({
127 pullId,
128 headOid: headOid as string,
129 baseOid: baseOid as string,
130 treeOid: treeOid as string,
131 commitOid: commitOid as string,
132 });
133 }
134 return steps;
135 }
136
137 function verdictOf(text: string): Judged | undefined {
138 const v = parsed(text);
139 if (typeof v !== "object" || v === null) return undefined;
140 const { lands, reopen, comment, learn, reason } = v as Record<string, unknown>;
141 if (lands === true) return { lands: true };
142 if (lands !== false || reopen !== true || typeof comment !== "string") return undefined;
143 if (!Array.isArray(learn) || !learn.every((f) => typeof f === "string")) return undefined;
144 const why = reasonOf(reason);
145 if (reason !== undefined && why === undefined) return undefined;
146 return { lands: false, reopen: true, comment, learn: learn as string[], ...(why === undefined ? {} : { reason: why }) };
147 }
148
149 /** Why a recorded red went red (#5276), read back; undefined when it is not one. */
150 function reasonOf(v: unknown): GateReason | undefined {
151 if (typeof v !== "object" || v === null) return undefined;
152 const { kind, failed } = v as Record<string, unknown>;
153 if (kind === "tests") return Number.isInteger(failed) && (failed as number) >= 0 ? { kind, failed: failed as number } : undefined;
154 if (kind === "typecheck" || kind === "check" || kind === "install" || kind === "setup" || kind === "timeout") return { kind };
155 return undefined;
156 }
157
158 /** Mark a pull red at a head and base, in place of any mark it had; undefined clears it. */
159 export function keepRed(
160 db: Database,
161 repoId: string,
162 pullId: string,
163 mark: Remembered | undefined,
164 now = new Date(),
165 ): void {
166 db.query("DELETE FROM queue_red WHERE pull_id = ?1").run(pullId);
167 if (mark === undefined) return;
168 db.query(
169 "INSERT INTO queue_red (repo_id, pull_id, head_oid, base_oid, together, at) VALUES (?1, ?2, ?3, ?4, ?5, ?6)",
170 ).run(repoId, pullId, mark.head, mark.base, JSON.stringify(mark.together), now.toISOString());
171 }
172
173 /** The repository's red marks, by pull id. */
174 export function redMarks(db: Database, repoId: string): Map<string, Remembered> {
175 const rows = db
176 .query<{ pull_id: string; head_oid: string; base_oid: string; together: string }, [string]>(
177 "SELECT pull_id, head_oid, base_oid, together FROM queue_red WHERE repo_id = ?1 ORDER BY at",
178 )
179 .all(repoId);
180 const marks = new Map<string, Remembered>();
181 for (const row of rows) {
182 const together = parsed(row.together);
183 const numbers =
184 Array.isArray(together) && together.every((n) => Number.isSafeInteger(n)) ? (together as number[]) : [];
185 marks.set(row.pull_id, { head: row.head_oid, base: row.base_oid, together: numbers });
186 }
187 return marks;
188 }
189
190 /**
191 * Whether the check lane is running a gate of `tipOid` now, in any process
192 * on this database: its `check_gate` trace is open and younger than `maxMs`
193 * (a trace a crash left open is no run).
194 */
195 export function gateRunning(db: Database, tipOid: string, maxMs: number, now = new Date()): boolean {
196 const since = new Date(now.getTime() - maxMs).toISOString();
197 return (
198 db
199 .query<{ n: number }, [string, string]>(
200 "SELECT COUNT(*) AS n FROM check_gate WHERE commit_oid = ?1 AND finished_at IS NULL AND started_at >= ?2",
201 )
202 .get(tipOid, since)!.n > 0
203 );
204 }
205
206 /** The verdict recorded for a gate, if it has one. */
207 export function recordedVerdict(db: Database, id: string): Judged | undefined {
208 const row = db.query<{ verdict: string | null }, [string]>("SELECT verdict FROM queue_gate WHERE id = ?1").get(id);
209 return row?.verdict == null ? undefined : verdictOf(row.verdict);
210 }
211
212 /** Whether a gate's record is still kept: gone once a line acted on it. */
213 export function gateRecorded(db: Database, id: string): boolean {
214 return db.query<{ id: string }, [string]>("SELECT id FROM queue_gate WHERE id = ?1").get(id) !== null;
215 }
src/checks/service.ts +83 −50
@@ -93,7 +93,7 @@ import { placementFromEnv, type Placement } from "./placement.ts";
9393 import { JOB_VANISHED, runJobStep } from "./job_step.ts";
9494 import { checkInstallCache, installWithRetry, withLock } from "./install.ts";
9595 import { checkStage, type GateStage } from "./gate_reason.ts";
96 import { extraGates, gateQueued, gatesInFlight, watchGates } from "./queue.ts";
96 import { extraGates, gateQueued, gateRunId, gatesInFlight, watchGates } from "./queue.ts";
9797 import type { JobOptions } from "../cluster/jobs.ts";
9898 import type { CheckDetail } from "../core/runs.ts";
9999
@@ -141,6 +141,15 @@ export interface ChecksOptions {
141141 readonly installRetryMs?: number | undefined;
142142 }
143143
144 /** A gate's answer from the lane: `interrupted` is no verdict (#3155). */
145 export interface GateDone {
146 ok: boolean;
147 error: string;
148 lines: string[];
149 interrupted?: boolean;
150 stage?: GateStage;
151 }
152
144153 export interface ChecksHandle {
145154 /** Ask for a check of one commit; the answer is the run, or nothing when the commit declares no check. */
146155 request(
@@ -159,12 +168,12 @@ export interface ChecksHandle {
159168 * forge stopping, or the cluster losing its Job — and is no verdict (#3155).
160169 * `signal` aborting (the queue's deadline, #3430) cuts it off the same way.
161170 */
162 gate(
163 repo: Repo,
164 pull: Pull,
165 commitOid: string,
166 signal?: AbortSignal,
167 ): Promise<{ ok: boolean; error: string; lines: string[]; interrupted?: boolean; stage?: GateStage }>;
171 gate(repo: Repo, pull: Pull, commitOid: string, signal?: AbortSignal): Promise<GateDone>;
172 /**
173 * The gate of `commitOid` this forge is running now, to wait on rather
174 * than start again (#6611); undefined when none is.
175 */
176 adopt?(commitOid: string): Promise<GateDone> | undefined;
168177 /**
169178 * Stop checking. Checks under way get `drainMs` to finish unless `drain`
170179 * is false (a person's Ctrl-C): then they are cut off and queued again now.
@@ -314,8 +323,12 @@ export function startChecks(app: App, options: ChecksOptions): ChecksHandle {
314323 capacityCount(app.db, "checks.concurrency", booted);
315324 const gate = options.gate ?? new Gate(capacity());
316325 const placement = options.placement ?? placementFromEnv();
317 /** Full gates on land in flight: not slots in the queue, but stopped with it. */
318 const gates = new Set<AbortController>();
326 /**
327 * Full gates on land in flight: not slots in the queue, but stopped with
328 * it — after the drain, as a check is. By commit, so a merge queue that
329 * carries on adopts one rather than gate the same tip again (#6611).
330 */
331 const gates = new Map<AbortController, { commitOid: string; done: Promise<GateDone> }>();
319332 const inFlight = new Map<
320333 string,
321334 { promise: Promise<void>; abort: AbortController }
@@ -1238,49 +1251,64 @@ export function startChecks(app: App, options: ChecksOptions): ChecksHandle {
12381251 schedule();
12391252 void catchUp();
12401253
1254 /** One gate's run on the lane: the full tier on `commitOid`, traced, cut off by `controller`. */
1255 async function gateOnce(
1256 repo: Repo,
1257 pull: Pull,
1258 commitOid: string,
1259 controller: AbortController,
1260 signal: AbortSignal | undefined,
1261 ): Promise<GateDone> {
1262 const lines: string[] = [];
1263 const abandon = (): void => controller.abort();
1264 signal?.addEventListener("abort", abandon, { once: true });
1265 // The slot minutes it spends are the flow's gate minutes (#6426).
1266 const traced = beginGate(app.db, repo.id, pull.id, commitOid);
1267 try {
1268 const bare = await pullDir(app.db, app.reposRoot, repo, pull);
1269 const done = await runTier({
1270 bare,
1271 repo,
1272 id: gateRunId(commitOid),
1273 commitOid,
1274 pull,
1275 tier: "full",
1276 write: (line) => lines.push(line),
1277 signal: controller.signal,
1278 });
1279 if (done.ok) return { ok: true, error: "", lines };
1280 const stage = done.stage ?? "setup";
1281 // Cut off by this forge stopping, the Job lost under it, or an
1282 // install the machine failed twice: the code was not judged, so
1283 // this is no red. A gate the drain lets finish is judged (#6611).
1284 if (
1285 controller.signal.aborted ||
1286 done.error === JOB_VANISHED ||
1287 done.infrastructure === true
1288 )
1289 return { ok: false, error: done.error, lines, interrupted: true, stage };
1290 return { ok: false, error: done.error, lines, stage };
1291 } finally {
1292 endGate(app.db, traced);
1293 signal?.removeEventListener("abort", abandon);
1294 gates.delete(controller);
1295 }
1296 }
1297
12411298 const handle: ChecksHandle = {
12421299 request,
1243 gate: async (repo, pull, commitOid, signal) => {
1244 const lines: string[] = [];
1300 gate: (repo, pull, commitOid, signal) => {
12451301 // A gate asked of a forge already stopping is not started: the forge
12461302 // that carries on asks it again.
1247 if (stopped) return { ok: false, error: INTERRUPTED, lines, interrupted: true };
1303 if (stopped) return Promise.resolve({ ok: false, error: INTERRUPTED, lines: [], interrupted: true });
12481304 const controller = new AbortController();
1249 gates.add(controller);
1250 const abandon = (): void => controller.abort();
1251 signal?.addEventListener("abort", abandon, { once: true });
1252 // The slot minutes it spends are the flow's gate minutes (#6426).
1253 const traced = beginGate(app.db, repo.id, pull.id, commitOid);
1254 try {
1255 const bare = await pullDir(app.db, app.reposRoot, repo, pull);
1256 const done = await runTier({
1257 bare,
1258 repo,
1259 id: `gate-${commitOid.slice(0, 12)}`,
1260 commitOid,
1261 pull,
1262 tier: "full",
1263 write: (line) => lines.push(line),
1264 signal: controller.signal,
1265 });
1266 if (done.ok) return { ok: true, error: "", lines };
1267 const stage = done.stage ?? "setup";
1268 // Cut off by this forge stopping, the Job lost under it, or an
1269 // install the machine failed twice: the code was not judged, so
1270 // this is no red.
1271 if (
1272 stopped ||
1273 controller.signal.aborted ||
1274 done.error === JOB_VANISHED ||
1275 done.infrastructure === true
1276 )
1277 return { ok: false, error: done.error, lines, interrupted: true, stage };
1278 return { ok: false, error: done.error, lines, stage };
1279 } finally {
1280 endGate(app.db, traced);
1281 signal?.removeEventListener("abort", abandon);
1282 gates.delete(controller);
1283 }
1305 const done = gateOnce(repo, pull, commitOid, controller, signal);
1306 gates.set(controller, { commitOid, done });
1307 return done;
1308 },
1309 adopt: (commitOid) => {
1310 for (const g of gates.values()) if (g.commitOid === commitOid) return g.done;
1311 return undefined;
12841312 },
12851313 spareSlots: () => Math.max(0, capacity() - PULL_RESERVE - inFlight.size - extraGates()),
12861314 lane: gate,
@@ -1291,9 +1319,14 @@ export function startChecks(app: App, options: ChecksOptions): ChecksHandle {
12911319 unwatch();
12921320 // A roll waits for the checks under way, up to the budget, and only
12931321 // then cuts off what is left (it is queued again, not lost).
1294 const running = Promise.all([...inFlight.values()].map((f) => f.promise));
1322 // A merge-queue gate is under way too: its verdict, if it comes in the
1323 // drain, is kept for the forge that carries on (#6611).
1324 const running = Promise.all([
1325 ...[...inFlight.values()].map((f) => f.promise),
1326 ...[...gates.values()].map((g) => g.done),
1327 ]);
12951328 const drainMs = stopOptions?.drain === false ? 0 : (options.drainMs ?? 0);
1296 if (drainMs > 0 && inFlight.size > 0) {
1329 if (drainMs > 0 && inFlight.size + gates.size > 0) {
12971330 let timer: ReturnType<typeof setTimeout> | undefined;
12981331 await Promise.race([
12991332 running,
@@ -1304,7 +1337,7 @@ export function startChecks(app: App, options: ChecksOptions): ChecksHandle {
13041337 clearTimeout(timer);
13051338 }
13061339 for (const { abort } of inFlight.values()) abort.abort();
1307 for (const abort of gates) abort.abort();
1340 for (const abort of gates.keys()) abort.abort();
13081341 await running;
13091342 },
13101343 };
src/core/migrations/20261026215105-queue-gate.ts added +30
@@ -0,0 +1,30 @@
1 // The merge queue survives a roll (#6611): each gate's candidate as it was
2 // built — its steps' commits and trees, the base, the tip the check lane
3 // runs as `run_id`, since when — and its verdict once it has one, so the
4 // forge that carries on resumes that very candidate instead of planning the
5 // line again. `queue_red` keeps the queue's red marks by pull, head and base
6 // (`checks/queue_store.ts`). Two new tables.
7 export default `
8 CREATE TABLE IF NOT EXISTS queue_gate (
9 id TEXT PRIMARY KEY,
10 repo_id TEXT NOT NULL REFERENCES repo(id) ON DELETE CASCADE,
11 run_id TEXT NOT NULL,
12 base_oid TEXT NOT NULL,
13 base_tree_oid TEXT NOT NULL,
14 tip_oid TEXT NOT NULL,
15 tree_oid TEXT NOT NULL,
16 steps TEXT NOT NULL,
17 started_at TEXT NOT NULL,
18 verdict TEXT
19 );
20 CREATE INDEX IF NOT EXISTS queue_gate_repo ON queue_gate (repo_id, started_at);
21 CREATE TABLE IF NOT EXISTS queue_red (
22 repo_id TEXT NOT NULL REFERENCES repo(id) ON DELETE CASCADE,
23 pull_id TEXT NOT NULL REFERENCES issue(id) ON DELETE CASCADE,
24 head_oid TEXT NOT NULL,
25 base_oid TEXT NOT NULL,
26 together TEXT NOT NULL DEFAULT '[]',
27 at TEXT NOT NULL,
28 PRIMARY KEY (pull_id, head_oid, base_oid)
29 );
30 `;
src/crew/autonomy.ts +38 −6
@@ -22,9 +22,10 @@ import {
2222 setIssueState,
2323 } from "../actions/issues.ts";
2424 import type { AccessError } from "../actions/access.ts";
25 import { gateCandidate, releaseOnLand } from "../checks/landing.ts";
25 import { adoptGate, gateCandidate, releaseOnLand } from "../checks/landing.ts";
26 import { queueStore } from "../checks/queue_store.ts";
2627 import { addLearned } from "../checks/learned.ts";
27 import { enqueue, type GateCandidate, type Queued, type QueueOptions, queueClashes, queueWords } from "../checks/queue.ts";
28 import { dropFromLine, enqueue, type GateCandidate, type Queued, type QueueOptions, queueClashes, queueWords, waitingPulls } from "../checks/queue.ts";
2829 import { checksFor } from "../checks/service.ts";
2930 import { queueDepth } from "../core/capacity.ts";
3031 import { readConflictLane, readConflictRoom } from "./conflict_room.ts";
@@ -1123,10 +1124,35 @@ export async function sweepGreen(app: App, repo: Repo): Promise<number> {
11231124 }
11241125 noteWaits(repo.id, tallyWaits(waits));
11251126 if (entries.length === 0) return 0;
1126 const landed = await enqueue(repo.id, entries, gateFor(app, repo), queueOptions(app));
1127 const landed = await enqueue(repo.id, entries, gateFor(app, repo), queueOptions(app, repo));
11271128 return landed.filter(Boolean).length;
11281129 }
11291130
1131 /**
1132 * Take out of the merge queue's line every waiting pull that would no
1133 * longer be admitted (#6611): held, closed or merged, pushed to since it
1134 * joined, no longer green on its head, or with changes requested on it.
1135 * Run on every tick, beside the sweep: a sweep in flight waits on its whole
1136 * line, so the line is cleared here, not there. Reads only rows: cheap.
1137 * How many were taken out.
1138 */
1139 export function pruneLine(app: App, repo: Repo): number {
1140 const out = new Set<string>();
1141 for (const joined of waitingPulls(repo.id)) {
1142 const row = getPull(app.db, repo.id, joined.number);
1143 const stays =
1144 row !== undefined &&
1145 row.state === "open" &&
1146 row.mergedAt === null &&
1147 row.headOid === joined.headOid &&
1148 getHold(app.db, row.id) === undefined &&
1149 verdictForHead(app.db, row.id, row.headOid).every((c) => c.state === "ok") &&
1150 reviewSummary(app.db, row.id, row.headOid).changesRequested === 0;
1151 if (!stays) out.add(joined.id);
1152 }
1153 return out.size === 0 ? 0 : dropFromLine(repo.id, out);
1154 }
1155
11301156 /** The sweep each repository has in flight, by repository id. */
11311157 const sweeping = new Map<string, Promise<number>>();
11321158
@@ -1211,6 +1237,9 @@ export async function autonomyTick(
12111237 // An archived repository is put away: the crew starts nothing in it.
12121238 if (repo.archivedAt !== null) continue;
12131239 try {
1240 // A hold, a close, a push or a lost approval leaves the line before
1241 // the next batch is formed, not at its gate (#6611).
1242 pruneLine(app, repo);
12141243 void startSweep(app, repo, sweep);
12151244 // First of all: a slot freed goes to an approved pull that no longer
12161245 // merges, before a changes-requested one or a new issue (#3682).
@@ -1798,7 +1827,7 @@ export async function landIfGreen(
17981827 ): Promise<boolean> {
17991828 const admitted = await admit(app, repo, pull);
18001829 if (!admitted.ok) return false;
1801 const [landed] = await enqueue(repo.id, [admitted.value], gateFor(app, repo), queueOptions(app));
1830 const [landed] = await enqueue(repo.id, [admitted.value], gateFor(app, repo), queueOptions(app, repo));
18021831 return landed === true;
18031832 }
18041833
@@ -2026,13 +2055,16 @@ export function regateDue(red: Decision, now: number): boolean {
20262055 /**
20272056 * How the merge queue may use the forge's check slots (#4053): its depth as
20282057 * an admin set it, and the slots the checks service has spare. Each judged
2029 * batch is kept for the red-batch rate (#6427).
2058 * batch is kept for the red-batch rate (#6427); its gates and red marks are
2059 * kept in the database, and a gate a roll left running adopted (#6611).
20302060 */
2031 function queueOptions(app: App): QueueOptions {
2061 function queueOptions(app: App, repo: Repo): QueueOptions {
20322062 return {
20332063 depth: queueDepth(app.db),
20342064 room: () => checksFor(app)?.spareSlots?.() ?? 0,
20352065 onBatch: (repoId, red) => keepBatch(app.db, repoId, red),
2066 store: queueStore(app.db),
2067 adopt: (record, signal) => adoptGate(app, repo, record, signal),
20362068 };
20372069 }
20382070
src/main.ts +4
@@ -34,6 +34,7 @@ import { AppsRuntime } from "./apps/runtime.ts";
3434 import { resolveAppDomain } from "./apps/host.ts";
3535 import { LAND_SETTLE_MS, requeueAbandonedChecks, startChecks } from "./checks/service.ts";
3636 import { RED_HOLD_MS } from "./checks/red_main.ts";
37 import { stopQueue } from "./checks/queue.ts";
3738 import { startLooks } from "./looks/service.ts";
3839 import { startMirrorOut } from "./origins/mirror-out.ts";
3940 import { startProbes } from "./contract/probes.ts";
@@ -657,6 +658,9 @@ function drain(signal: NodeJS.Signals): Promise<unknown> {
657658 // The platform's SIGTERM: the listener stays open while routers learn
658659 // this process is going. A person's Ctrl-C does not wait for routers.
659660 const linger = signal === "SIGTERM" ? Bun.sleep(lingerMs ?? 0) : undefined;
661 // The merge queue acts on nothing more: its gates run on through the
662 // checks' drain, and a verdict is kept for the forge that carries on (#6611).
663 stopQueue();
660664 return Promise.all([
661665 linger,
662666 prompts.stop(),
test/gate-drain.test.ts added +95
@@ -0,0 +1,95 @@
1 /**
2 * A roll lets a merge-queue gate under way finish in the checks' drain
3 * (#6611), as it lets a check finish: its verdict is judged, not cut off,
4 * so the queue keeps it for the forge that carries on. While it runs, the
5 * lane hands the same run to anyone who asks to adopt its commit.
6 */
7 import { afterAll, beforeAll, describe, expect, test } from "bun:test";
8 import { mkdtemp, rm } from "node:fs/promises";
9 import { tmpdir } from "node:os";
10 import { join } from "node:path";
11 import { createApp, type App } from "../src/app.ts";
12 import { openPull } from "../src/actions/pulls.ts";
13 import { commitToBranch, createRepo } from "../src/actions/repos.ts";
14 import { gateCandidate } from "../src/checks/landing.ts";
15 import { startChecks, type ChecksHandle } from "../src/checks/service.ts";
16 import { getPull } from "../src/core/pulls.ts";
17 import type { Repo } from "../src/core/repos.ts";
18 import { createUser, type User } from "../src/core/users.ts";
19 import { repoDir } from "../src/git/repos.ts";
20
21 describe("a gate under way when the forge rolls", () => {
22 let dataDir: string;
23 let app: App;
24 let checks: ChecksHandle;
25 let repo: Repo;
26 let alice: User;
27 const cells = process.env["FORGE_CELLS"];
28
29 beforeAll(async () => {
30 process.env["FORGE_CELLS"] = "off";
31 dataDir = await mkdtemp(join(tmpdir(), "forge-gate-drain-"));
32 app = createApp({ dataDir, port: 0 });
33 checks = startChecks(app, { forgeUrl: "http://127.0.0.1:1", concurrency: 1, drainMs: 60_000 });
34 const a = await createUser(app.db, {
35 origin: "seed",
36 trusted: true,
37 username: "alice",
38 email: "[email protected]",
39 password: "a-good-passphrase",
40 adminIfFirst: true,
41 });
42 if (!a.ok) throw new Error(a.error);
43 alice = a.value;
44 const made = await createRepo(app, alice, { name: "drain", visibility: "public" });
45 if (!made.ok) throw new Error(String(made.error));
46 repo = made.value;
47 const seeded = await commitToBranch(app, alice, repo, {
48 message: "Start",
49 files: [
50 {
51 path: "forge.json",
52 content: JSON.stringify({
53 check: { run: "true", fast: "true", full: "sleep 1; echo test/slow.test.ts:; exit 1" },
54 }),
55 },
56 ],
57 });
58 if (!seeded.ok) throw new Error(JSON.stringify(seeded.error));
59 const branched = await commitToBranch(app, alice, repo, {
60 branch: "feature",
61 message: "Touch b",
62 files: [{ path: "b.txt", content: "b\n" }],
63 });
64 if (!branched.ok) throw new Error(JSON.stringify(branched.error));
65 });
66
67 afterAll(async () => {
68 await checks.stop({ drain: false });
69 app.close();
70 if (cells === undefined) delete process.env["FORGE_CELLS"];
71 else process.env["FORGE_CELLS"] = cells;
72 await rm(dataDir, { recursive: true, force: true });
73 });
74
75 test("the drain waits for it, its red is judged, and the lane hands its run to an adopter", async () => {
76 const opened = await openPull(app, alice, repo, { title: "Feature", baseBranch: "main", headBranch: "feature" });
77 if (!opened.ok) throw new Error(String(opened.error));
78 const pull = getPull(app.db, repo.id, opened.value.number)!;
79 const dir = repoDir(app.reposRoot, repo.ownerName, repo.name);
80 expect(checks.adopt?.(pull.headOid)).toBeUndefined();
81 const gating = gateCandidate(app, repo, pull, dir, pull.headOid);
82 let adopted = checks.adopt?.(pull.headOid);
83 for (let i = 0; i < 400 && adopted === undefined; i++) {
84 await Bun.sleep(5);
85 adopted = checks.adopt?.(pull.headOid);
86 }
87 expect(adopted).toBeDefined();
88 await checks.stop({ drain: true });
89 const verdict = await gating;
90 expect(verdict.lands).toBe(false);
91 expect("reopen" in verdict && verdict.reopen).toBe(true);
92 expect((await adopted!).interrupted).toBeUndefined();
93 expect(checks.adopt?.(pull.headOid)).toBeUndefined();
94 }, 60_000);
95 });
test/merge-queue-hold.test.ts added +172
@@ -0,0 +1,172 @@
1 /**
2 * A hold takes a pull out of the merge queue's line (#6611), not just out of
3 * the next admission: a pull waiting behind a gate that is then held, closed,
4 * pushed to or sent back for changes leaves the line on the next tick, the
5 * tick line's waiting count says so, and the next batch is formed without it.
6 */
7 import { afterAll, beforeAll, describe, expect, test } from "bun:test";
8 import { mkdtemp, rm } from "node:fs/promises";
9 import { tmpdir } from "node:os";
10 import { join } from "node:path";
11 import { createApp, type App } from "../src/app.ts";
12 import { openPull } from "../src/actions/pulls.ts";
13 import { commitToBranch, createRepo } from "../src/actions/repos.ts";
14 import { queueWords, stopQueue } from "../src/checks/queue.ts";
15 import { gateRecords } from "../src/checks/queue_store.ts";
16 import { provideChecksForTest, type ChecksHandle, type GateDone } from "../src/checks/service.ts";
17 import { setHold } from "../src/core/holds.ts";
18 import { getPull, type Pull } from "../src/core/pulls.ts";
19 import { getRepoById, type Repo } from "../src/core/repos.ts";
20 import { claimRun, createRun, finishRun } from "../src/core/runs.ts";
21 import { createUser, type User } from "../src/core/users.ts";
22 import { landIfGreen, pruneLine } from "../src/crew/autonomy.ts";
23
24 describe("a hold and the merge queue's line", () => {
25 let dataDir: string;
26 let app: App;
27 let alice: User;
28 /** Gates running on the stand-in lane, by commit: what `adopt` finds. */
29 const running = new Map<string, { done: Promise<GateDone>; go: (done: GateDone) => void }>();
30 /** Every gate the lane was asked to start, by commit. */
31 let started: string[] = [];
32 /** The lane lost its runs (the pod went with them): nothing to adopt. */
33 let lost = false;
34
35
36 beforeAll(async () => {
37 dataDir = await mkdtemp(join(tmpdir(), "forge-merge-queue-hold-"));
38 app = createApp({ dataDir, port: 0 });
39 const a = await createUser(app.db, {
40 origin: "seed",
41 trusted: true,
42 username: "alice",
43 email: "[email protected]",
44 password: "a-good-passphrase",
45 adminIfFirst: true,
46 });
47 if (!a.ok) throw new Error(a.error);
48 alice = a.value;
49 provideChecksForTest(app, {
50 gate: (_repo: Repo, _pull: Pull, commitOid: string) => {
51 started.push(commitOid);
52 let go!: (done: GateDone) => void;
53 const done = new Promise<GateDone>((resolve) => (go = resolve));
54 running.set(commitOid, { done, go });
55 void done.then(() => running.delete(commitOid));
56 return done;
57 },
58 adopt: (commitOid: string) => (lost ? undefined : running.get(commitOid)?.done),
59 spareSlots: () => 0,
60 } as unknown as ChecksHandle);
61 });
62
63 afterAll(async () => {
64 stopQueue();
65 for (const r of running.values()) r.go({ ok: false, error: "", lines: [], interrupted: true });
66 await app.close?.();
67 await rm(dataDir, { recursive: true, force: true });
68 });
69
70 async function autonomousRepo(name: string): Promise<Repo> {
71 const made = await createRepo(app, alice, { name, visibility: "private" });
72 if (!made.ok) throw new Error(String(made.error));
73 const seeded = await commitToBranch(app, alice, made.value, {
74 message: "Start",
75 files: [
76 { path: "README.md", content: `# ${name}\n` },
77 { path: "forge.json", content: JSON.stringify({ check: { run: "true", fast: "true", full: "true" } }) },
78 ],
79 });
80 if (!seeded.ok) throw new Error(JSON.stringify(seeded.error));
81 app.db.query("UPDATE repo SET autonomy = 'autonomous', autonomy_by = ?2 WHERE id = ?1").run(made.value.id, alice.id);
82 return getRepoById(app.db, made.value.id)!;
83 }
84
85 async function greenPull(repo: Repo, path: string): Promise<Pull> {
86 const pushed = await commitToBranch(app, alice, repo, {
87 branch: `feature/${path}`,
88 base: "main",
89 message: path,
90 files: [{ path, content: `export const n = "${path}";\n` }],
91 });
92 if (!pushed.ok) throw new Error(JSON.stringify(pushed.error));
93 const opened = await openPull(app, alice, repo, { title: path, baseBranch: "main", headBranch: `feature/${path}` });
94 if (!opened.ok) throw new Error(String(opened.error));
95 const pull = getPull(app.db, repo.id, opened.value.number)!;
96 const check = createRun(app.db, {
97 repoId: repo.id,
98 kind: "check",
99 issueId: pull.id,
100 env: `pull-${pull.number}`,
101 commitOid: pull.headOid,
102 reason: "pull",
103 });
104 claimRun(app.db, check.id);
105 finishRun(app.db, check.id, { state: "ok", summary: "check passed" });
106 return pull;
107 }
108
109 async function gating(n: number): Promise<void> {
110 for (let i = 0; i < 2000 && running.size < n; i++) await Bun.sleep(5);
111 expect(running.size).toBe(n);
112 }
113
114 /** Each pull joins the line after the one before it has: the order is the test's. */
115 async function joinInTurn(repo: Repo, pulls: readonly Pull[]): Promise<Promise<boolean>[]> {
116 const landing: Promise<boolean>[] = [];
117 for (const [i, pull] of pulls.entries()) {
118 landing.push(landIfGreen(app, repo, pull));
119 for (let n = 0; n < 4000 && !queueWords(repo.id).includes(`${i + 1} waiting`); n++) await Bun.sleep(5);
120 expect(queueWords(repo.id)).toContain(`${i + 1} waiting`);
121 }
122 return landing;
123 }
124
125 test("three pulls wait, the middle one is held: the line says 2 waiting and the next batch has 2", async () => {
126 const repo = await autonomousRepo("hold-middle");
127 const first = await greenPull(repo, "first.ts");
128 const landingFirst = landIfGreen(app, repo, first);
129 await gating(1);
130 const busy = started.at(-1)!;
131 const [a, b, c] = [await greenPull(repo, "a.ts"), await greenPull(repo, "b.ts"), await greenPull(repo, "c.ts")];
132 const landing = await joinInTurn(repo, [a, b, c]);
133 expect(queueWords(repo.id)).toContain("3 waiting");
134 setHold(app.db, b.id, alice.id, "not this one yet");
135 expect(pruneLine(app, repo)).toBe(1);
136 expect(queueWords(repo.id)).toContain("2 waiting");
137 expect(await landing[1]).toBe(false);
138 // The gate ahead answers: the next batch is formed of the two left.
139 running.get(busy)!.go({ ok: true, error: "", lines: [] });
140 expect(await landingFirst).toBe(true);
141 await gating(1);
142 const [next] = gateRecords(app.db, repo.id);
143 expect(next!.steps.map((s) => s.pullId)).toEqual([a.id, c.id]);
144 running.get(started.at(-1)!)!.go({ ok: true, error: "", lines: [] });
145 expect(await landing[0]).toBe(true);
146 expect(await landing[2]).toBe(true);
147 expect(getPull(app.db, repo.id, b.number)!.mergedAt).toBeNull();
148 }, 60_000);
149
150 test("a pull pushed to or sent back for changes leaves the line; a pull still fit stays", async () => {
151 const repo = await autonomousRepo("hold-others");
152 const first = await greenPull(repo, "first.ts");
153 const landingFirst = landIfGreen(app, repo, first);
154 await gating(1);
155 const busy = started.at(-1)!;
156 const [a, b, c] = [await greenPull(repo, "a.ts"), await greenPull(repo, "b.ts"), await greenPull(repo, "c.ts")];
157 const landing = await joinInTurn(repo, [a, b, c]);
158 // a is pushed to; c is closed. b stays.
159 app.db.query("UPDATE pull SET head_oid = ?2 WHERE issue_id = ?1").run(a.id, "f".repeat(40));
160 app.db.query("UPDATE issue SET state = 'closed' WHERE id = ?1").run(c.id);
161 expect(pruneLine(app, repo)).toBe(2);
162 expect(queueWords(repo.id)).toContain("1 waiting");
163 expect(await landing[0]).toBe(false);
164 expect(await landing[2]).toBe(false);
165 expect(pruneLine(app, repo)).toBe(0);
166 running.get(busy)!.go({ ok: true, error: "", lines: [] });
167 expect(await landingFirst).toBe(true);
168 await gating(1);
169 running.get(started.at(-1)!)!.go({ ok: true, error: "", lines: [] });
170 expect(await landing[1]).toBe(true);
171 }, 60_000);
172 });
test/merge-queue-restart.test.ts added +301
@@ -0,0 +1,301 @@
1 /**
2 * The merge queue survives a roll (#6611). A gate's candidate is recorded
3 * as it starts; a queue stopped mid-gate and started again resumes that
4 * very candidate — the run still going on the check lane adopted, a verdict
5 * that came in after the stop acted on, or, the run gone, the same tip
6 * gated again on the same base — and never plans the line from nothing.
7 * The lane here is a stand-in the test holds open and lets go of.
8 */
9 import { afterAll, beforeAll, describe, expect, test } from "bun:test";
10 import { mkdtemp, rm } from "node:fs/promises";
11 import { tmpdir } from "node:os";
12 import { join } from "node:path";
13 import { createApp, type App } from "../src/app.ts";
14 import { openPull } from "../src/actions/pulls.ts";
15 import { commitToBranch, createRepo } from "../src/actions/repos.ts";
16 import { beginGate, endGate } from "../src/checks/gate_record.ts";
17 import { adoptGate } from "../src/checks/landing.ts";
18 import { type GateRecord, gateRuns, stopQueue } from "../src/checks/queue.ts";
19 import { beginGateRecord, gateRecords, keepRed, queueStore, redMarks, settleGateRecord } from "../src/checks/queue_store.ts";
20 import { provideChecksForTest, type ChecksHandle, type GateDone } from "../src/checks/service.ts";
21 import { getPull, type Pull } from "../src/core/pulls.ts";
22 import { getRepoById, type Repo } from "../src/core/repos.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 { repoDir } from "../src/git/repos.ts";
27
28 describe("the merge queue across a restart", () => {
29 let dataDir: string;
30 let app: App;
31 let alice: User;
32 /** Gates running on the stand-in lane, by commit: what `adopt` finds. */
33 const running = new Map<string, { done: Promise<GateDone>; go: (done: GateDone) => void }>();
34 /** Every gate the lane was asked to start, by commit. */
35 let started: string[] = [];
36 /** The lane lost its runs (the pod went with them): nothing to adopt. */
37 let lost = false;
38
39 const git = (repo: Repo, ...args: string[]) =>
40 Bun.spawnSync(["git", ...args], { cwd: repoDir(app.reposRoot, repo.ownerName, repo.name) })
41 .stdout.toString()
42 .trim();
43
44 beforeAll(async () => {
45 dataDir = await mkdtemp(join(tmpdir(), "forge-merge-queue-restart-"));
46 app = createApp({ dataDir, port: 0 });
47 const a = await createUser(app.db, {
48 origin: "seed",
49 trusted: true,
50 username: "alice",
51 email: "[email protected]",
52 password: "a-good-passphrase",
53 adminIfFirst: true,
54 });
55 if (!a.ok) throw new Error(a.error);
56 alice = a.value;
57 provideChecksForTest(app, {
58 gate: (_repo: Repo, _pull: Pull, commitOid: string) => {
59 started.push(commitOid);
60 let go!: (done: GateDone) => void;
61 const done = new Promise<GateDone>((resolve) => (go = resolve));
62 running.set(commitOid, { done, go });
63 void done.then(() => running.delete(commitOid));
64 return done;
65 },
66 adopt: (commitOid: string) => (lost ? undefined : running.get(commitOid)?.done),
67 spareSlots: () => 0,
68 } as unknown as ChecksHandle);
69 });
70
71 afterAll(async () => {
72 stopQueue();
73 for (const r of running.values()) r.go({ ok: false, error: "", lines: [], interrupted: true });
74 await app.close?.();
75 await rm(dataDir, { recursive: true, force: true });
76 });
77
78 async function autonomousRepo(name: string): Promise<Repo> {
79 const made = await createRepo(app, alice, { name, visibility: "private" });
80 if (!made.ok) throw new Error(String(made.error));
81 const seeded = await commitToBranch(app, alice, made.value, {
82 message: "Start",
83 files: [
84 { path: "README.md", content: `# ${name}\n` },
85 { path: "forge.json", content: JSON.stringify({ check: { run: "true", fast: "true", full: "true" } }) },
86 ],
87 });
88 if (!seeded.ok) throw new Error(JSON.stringify(seeded.error));
89 app.db.query("UPDATE repo SET autonomy = 'autonomous', autonomy_by = ?2 WHERE id = ?1").run(made.value.id, alice.id);
90 return getRepoById(app.db, made.value.id)!;
91 }
92
93 async function greenPull(repo: Repo, path: string): Promise<Pull> {
94 const pushed = await commitToBranch(app, alice, repo, {
95 branch: `feature/${path}`,
96 base: "main",
97 message: path,
98 files: [{ path, content: `export const n = "${path}";\n` }],
99 });
100 if (!pushed.ok) throw new Error(JSON.stringify(pushed.error));
101 const opened = await openPull(app, alice, repo, { title: path, baseBranch: "main", headBranch: `feature/${path}` });
102 if (!opened.ok) throw new Error(String(opened.error));
103 const pull = getPull(app.db, repo.id, opened.value.number)!;
104 const check = createRun(app.db, {
105 repoId: repo.id,
106 kind: "check",
107 issueId: pull.id,
108 env: `pull-${pull.number}`,
109 commitOid: pull.headOid,
110 reason: "pull",
111 });
112 claimRun(app.db, check.id);
113 finishRun(app.db, check.id, { state: "ok", summary: "check passed" });
114 return pull;
115 }
116
117 async function gating(n: number): Promise<void> {
118 for (let i = 0; i < 2000 && running.size < n; i++) await Bun.sleep(5);
119 expect(running.size).toBe(n);
120 }
121
122 /** Three green pulls at the gate as one candidate, recorded; then the queue is killed. */
123 async function killedMidGate(name: string) {
124 const repo = await autonomousRepo(name);
125 const pulls = [await greenPull(repo, "a.ts"), await greenPull(repo, "b.ts"), await greenPull(repo, "c.ts")];
126 const base = git(repo, "rev-parse", "main");
127 started = [];
128 lost = false;
129 const first = sweepGreen(app, repo);
130 await gating(1);
131 const [tip] = started;
132 const [record] = gateRecords(app.db, repo.id);
133 expect(record).toBeDefined();
134 expect(record!.tipOid).toBe(tip!);
135 expect(record!.baseOid).toBe(base);
136 expect(record!.runId).toBe(`gate-${tip!.slice(0, 12)}`);
137 expect(record!.steps.map((s) => s.pullId)).toEqual(pulls.map((p) => p.id));
138 stopQueue();
139 // What waited on the killed queue is answered: nothing landed.
140 expect(await first).toBe(0);
141 return { repo, pulls, base, tip: tip!, record: record! };
142 }
143
144 const merged = (repo: Repo, pull: Pull) => getPull(app.db, repo.id, pull.number)!.mergedAt !== null;
145
146 test("killed mid-gate and started again, the same candidate lands with no second gate run", async () => {
147 const { repo, pulls, base, tip, record } = await killedMidGate("restart-adopt");
148 const again = sweepGreen(app, repo);
149 // The new line adopts the run still going on the lane: none is started.
150 for (let i = 0; i < 20; i++) await Bun.sleep(5);
151 expect(started).toEqual([tip]);
152 running.get(tip)!.go({ ok: true, error: "", lines: [] });
153 expect(await again).toBe(3);
154 expect(started).toEqual([tip]);
155 expect(gateRuns(repo.id)).toBe(0);
156 for (const pull of pulls) expect(merged(repo, pull)).toBe(true);
157 // Main is exactly the tree that was gated, on the base it was built on.
158 expect(git(repo, "rev-parse", "main^{tree}")).toBe(record.treeOid);
159 expect(git(repo, "rev-list", "--count", `${base}..main`)).toBe("3");
160 expect(gateRecords(app.db, repo.id)).toEqual([]);
161 });
162
163 test("a verdict that came in after the stop is acted on, with no gate at all", async () => {
164 const { repo, pulls, tip, record } = await killedMidGate("restart-verdict");
165 // The stopped forge's gate answers on its drain: kept in the record.
166 running.get(tip)!.go({ ok: true, error: "", lines: [] });
167 for (let i = 0; i < 200 && gateRecords(app.db, repo.id)[0]?.verdict === undefined; i++) await Bun.sleep(5);
168 expect(gateRecords(app.db, repo.id)[0]?.verdict).toEqual({ lands: true });
169 lost = true;
170 expect(await sweepGreen(app, repo)).toBe(3);
171 expect(started).toEqual([tip]);
172 for (const pull of pulls) expect(merged(repo, pull)).toBe(true);
173 expect(git(repo, "rev-parse", "main^{tree}")).toBe(record.treeOid);
174 });
175
176 test("the run gone, the same candidate is gated again on the same base, not planned again", async () => {
177 const { repo, pulls, tip, record } = await killedMidGate("restart-regate");
178 // The forge stopping cut its gate off: no verdict, and the record stays.
179 running.get(tip)!.go({ ok: false, error: "cut off", lines: [], interrupted: true });
180 await Bun.sleep(20);
181 expect(gateRecords(app.db, repo.id).map((r) => r.id)).toEqual([record.id]);
182 lost = true;
183 const again = sweepGreen(app, repo);
184 await gating(1);
185 // The very commit that was gated before, not a candidate built anew.
186 expect(started).toEqual([tip, tip]);
187 running.get(tip)!.go({ ok: true, error: "", lines: [] });
188 expect(await again).toBe(3);
189 for (const pull of pulls) expect(merged(repo, pull)).toBe(true);
190 expect(git(repo, "rev-parse", "main^{tree}")).toBe(record.treeOid);
191 });
192
193 test("a recorded candidate whose base moved is forgotten, and the line is planned afresh", async () => {
194 const { repo, pulls, tip } = await killedMidGate("restart-moved");
195 running.get(tip)!.go({ ok: false, error: "cut off", lines: [], interrupted: true });
196 const moved = await commitToBranch(app, alice, repo, {
197 message: "main moves",
198 files: [{ path: "other.ts", content: "export {};\n" }],
199 });
200 if (!moved.ok) throw new Error(JSON.stringify(moved.error));
201 const again = sweepGreen(app, repo);
202 await gating(1);
203 expect(started).toHaveLength(2);
204 expect(started[1]).not.toBe(tip);
205 running.get(started[1]!)!.go({ ok: true, error: "", lines: [] });
206 expect(await again).toBe(3);
207 for (const pull of pulls) expect(merged(repo, pull)).toBe(true);
208 expect(gateRecords(app.db, repo.id)).toEqual([]);
209 });
210
211 test("red marks are kept by pull, head and base, one per pull, and read back", async () => {
212 const repo = await autonomousRepo("restart-reds");
213 const pull = await greenPull(repo, "r.ts");
214 const head = pull.headOid;
215 const base = git(repo, "rev-parse", "main");
216 const store = queueStore(app.db);
217 store.red(repo.id, pull.id, { head, base, together: [4, 5] });
218 expect(redMarks(app.db, repo.id).get(pull.id)).toEqual({ head, base, together: [4, 5] });
219 // A new mark replaces the old: the line remembers one per pull.
220 keepRed(app.db, repo.id, pull.id, { head, base: head, together: [] });
221 expect([...redMarks(app.db, repo.id).values()]).toEqual([{ head, base: head, together: [] }]);
222 store.red(repo.id, pull.id, undefined);
223 expect(redMarks(app.db, repo.id).size).toBe(0);
224 });
225
226 test("a record that does not read is no record", async () => {
227 const repo = await autonomousRepo("restart-garbage");
228 app.db
229 .query(
230 `INSERT INTO queue_gate (id, repo_id, run_id, base_oid, base_tree_oid, tip_oid, tree_oid, steps, started_at)
231 VALUES ('x', ?1, 'gate-x', 'nope', 'nope', 'nope', 'nope', '{', ?2)`,
232 )
233 .run(repo.id, new Date().toISOString());
234 expect(gateRecords(app.db, repo.id)).toEqual([]);
235 });
236
237 test("a red kept across a roll keeps why it went red (#5276): an install red is not searched as the code's", async () => {
238 const repo = await autonomousRepo("restart-reason");
239 const pull = await greenPull(repo, "w.ts");
240 const base = git(repo, "rev-parse", "main");
241 const tree = git(repo, "rev-parse", "main^{tree}");
242 const tip = pull.headOid;
243 beginGateRecord(app.db, {
244 id: "reasoned-gate",
245 repoId: repo.id,
246 runId: `gate-${tip.slice(0, 12)}`,
247 baseOid: base,
248 baseTreeOid: tree,
249 tipOid: tip,
250 treeOid: tree,
251 steps: [{ pullId: pull.id, headOid: tip, baseOid: base, treeOid: tree, commitOid: tip }],
252 startedAt: new Date().toISOString(),
253 });
254 const red = { lands: false, reopen: true, comment: "red", learn: [], reason: { kind: "install" } } as const;
255 settleGateRecord(app.db, "reasoned-gate", red);
256 expect(gateRecords(app.db, repo.id)[0]?.verdict).toEqual(red);
257 app.db.query("DELETE FROM queue_gate WHERE id = 'reasoned-gate'").run();
258 });
259 test("a gate the forge before this one still runs is waited on: its verdict comes into the record", async () => {
260 const repo = await autonomousRepo("restart-draining");
261 const pull = await greenPull(repo, "d.ts");
262 const base = git(repo, "rev-parse", "main");
263 const tree = git(repo, "rev-parse", "main^{tree}");
264 const tip = pull.headOid;
265 const record: GateRecord = {
266 id: "draining-gate",
267 repoId: repo.id,
268 runId: `gate-${tip.slice(0, 12)}`,
269 baseOid: base,
270 baseTreeOid: tree,
271 tipOid: tip,
272 treeOid: tree,
273 steps: [{ pullId: pull.id, headOid: tip, baseOid: base, treeOid: tree, commitOid: tip }],
274 startedAt: new Date().toISOString(),
275 };
276 beginGateRecord(app.db, record);
277 // No run of it on this lane, and no trace of one anywhere: gone.
278 expect(await adoptGate(app, repo, record, new AbortController().signal, 5)).toBeUndefined();
279 // The other forge's lane holds it open; its verdict arrives in the record.
280 const trace = beginGate(app.db, repo.id, pull.id, tip);
281 const waited = adoptGate(app, repo, record, new AbortController().signal, 5);
282 await Bun.sleep(30);
283 settleGateRecord(app.db, record.id, { lands: true });
284 endGate(app.db, trace);
285 expect(await waited).toEqual({ lands: true });
286 // Its trace closed with no verdict: the run is gone.
287 const cut = beginGate(app.db, repo.id, pull.id, tip);
288 app.db.query("UPDATE queue_gate SET verdict = NULL WHERE id = ?1").run(record.id);
289 const gone = adoptGate(app, repo, record, new AbortController().signal, 5);
290 await Bun.sleep(20);
291 endGate(app.db, cut);
292 expect(await gone).toBeUndefined();
293 // The other forge's queue was not stopped: it acted on the verdict and
294 // forgot the record. Nothing is told here, and nothing gated again.
295 const live = beginGate(app.db, repo.id, pull.id, tip);
296 const elsewhere = adoptGate(app, repo, record, new AbortController().signal, 5);
297 queueStore(app.db).forget(record.id);
298 endGate(app.db, live);
299 expect(await elsewhere).toEqual({ lands: false, interrupted: true });
300 });
301 });

↑↓ navigate ↵ open ⌘↵ its app esc close

Keyboard shortcuts

Docs