1acd299 · 4e21dfe · a553e32 · 93c49ea
@@ -1087,6 +1087,10 @@
word) and `issue.parked = 1` carries it, so raw SQL never reads it as done.
its repository's queue (`src/checks/queue.ts`). The queue builds one
candidate: main, then each queued pull's squash on top of the one before, in
queue order, leaving out any pull that conflicts or has nothing left to land.
A candidate carries at most eight pulls (`MAX_CANDIDATE`), so a gate stays
near one full tier and a red bisects in three; the rest wait their turn.
The log says `[queue] gating N pulls since HH:MM` as a gate starts and
`[queue] gated N pulls: green|red|cut off` as it ends.
The full gate runs once on that candidate. Green lands every pull in order,
and each landing is refused unless main is still where the candidate was
built and the squash writes the tree that was gated, so main ends at the
@@ -1096,7 +1100,9 @@
waits for a push or a new main, except once: after 30 minutes with neither
side moved it is gated again, and lands if green; a red that reproduces
stays red and is not told twice. A gate cut off by the forge stopping (a
roll, a drain) or by the cluster losing its Job is no red at all: the next
sweep gates it again. Pulls
sweep gates it again. So is a gate that has not answered in 15 minutes
(`GATE_DEADLINE_MS`): it is abandoned and its check aborted, the log says
`[queue] candidate abandoned`, and the next sweep forms the line again. Pulls
that turn green while a candidate is gating wait for the next one. Each
queue's numbers — how many pulls are queued, how many are gating, and how
many landed in the last hour — show in three places: the flow panel on `/boards`, the
@@ -1110,7 +1116,9 @@
own first; the batch lands in one gate. A pull's own check runs again only
when its head changes. A person's merge, which no candidate gated, still
refuses a stale green. The autonomy tick line counts the approved pulls the
last sweep saw and why each waits (`approved="3 approved: queued=2 review=1"`,
from `src/crew/waits.ts`).
from `src/crew/waits.ts`), and what the queue is doing now
(`queue="gating 2 pulls since 18:54"`, `queue="1 waiting"` or `queue=idle`,
in UTC).
An autonomous app's planner is **held** while nothing is landing or the plate
is full: no pull has merged in 48 h while one was closed unmerged, or more than
@@ -35,6 +35,7 @@
import { ownProbeHeaders } from "../kernel/probe.ts";
 * pull with its base, or the merge queue's candidate of several (#2036).
 * `pull` is the one the gate is asked for: where the objects are and which
 * base branch's check is run. The manifest is the candidate's own.
 * `signal` is the queue abandoning the gate at its deadline (#3430).
 */
export async function gateCandidate(
  app: App,
@@ -42,6 +43,7 @@
export async function gateCandidate(
  pull: Pull,
  dir: string,
  commitOid: string,
  signal?: AbortSignal,
): Promise<LandVerdict> {
  const checks = checksFor(app);
  if (checks === undefined) return { lands: true };
@@ -52,7 +54,7 @@
export async function gateCandidate(
  const declared = parseCheck(forgeJson, packageJson);
  if (!declared.ok || declared.value === undefined || declared.value.fast === null)
    return { lands: true };
  const done = await checks.gate(repo, pull, commitOid);
  const done = await checks.gate(repo, pull, commitOid, signal);
  // Cut off, not judged: no verdict, and not the gate log a red reads (#3155).
  if (done.interrupted === true) return { lands: false, interrupted: true };
  keepForgeGate(app.config.dataDir, app.config.forgeRepo, `${repo.ownerName}/${repo.name}`, done.lines);
@@ -10,7 +10,10 @@
 * search in the first half while the rest waits for the next candidate. A
 * single red step is the culprit and is handed back. A gate cut off by the
 * forge stopping is no verdict: nothing is handed back, and the forge that
 * carries on admits the line again (#3155).
 * carries on admits the line again (#3155). A gate that has not answered
 * by `GATE_DEADLINE_MS` is abandoned the same way: cut off, logged, and the
 * line re-formed by the next sweep, so one gate that never returns cannot
 * hold the queue for good (#3430).
 *
 * What is queued is not stored: it is the pulls `landIfGreen` admitted, and
 * every sweep admits them again. The queue only orders them and remembers
@@ -67,8 +70,38 @@
export interface Candidate {
  readonly treeOid: string;
}
/** Gate a candidate: the full tier on its tip. */
export type GateCandidate = (candidate: Candidate) => Promise<LandVerdict>;
/**
 * Gate a candidate: the full tier on its tip. `signal` aborts when the
 * queue abandons the gate at its deadline: what it started should stop.
 */
export type GateCandidate = (candidate: Candidate, signal: AbortSignal) => Promise<LandVerdict>;
/**
 * How long a candidate's gate may run before the queue abandons it (#3430):
 * twice the forge's full tier. Past it the gate is no verdict, as a gate the
 * forge stopping cut off is not (#3155).
 */
export const GATE_DEADLINE_MS = 15 * 60_000;
let deadlineMs = GATE_DEADLINE_MS;
/**
 * The most pulls one candidate carries (#3430): eight keep a gate near one
 * full tier and a red's bisection three gates deep. The rest of the line
 * waits for the next candidate, in order.
 */
export const MAX_CANDIDATE = 8;
/** Tests shorten the deadline; the answer puts it back. */
export function gateDeadlineForTest(ms: number): () => void {
  deadlineMs = ms;
  return () => {
    deadlineMs = GATE_DEADLINE_MS;
  };
}
/** The gate as a round asks it: the deadline is the line's own. */
type Gated = (candidate: Candidate) => Promise<LandVerdict>;
interface Waiting {
  readonly entry: Queued;
@@ -76,11 +109,14 @@
interface Waiting {
}
interface Line {
  readonly repoId: string;
  waiting: Waiting[];
  /** Pull ids in the queue or its candidate: a pull is in the line once. */
  readonly holds: Set<string>;
  /** How many pulls the candidate at the gate now carries. */
  gating: number;
  /** When the candidate at the gate started, epoch ms; 0 when none is. */
  since: number;
  /** The gate runs this line has made, for a test and the log. */
  gates: number;
  draining: boolean;
@@ -100,7 +136,7 @@
const IDENTITY = {
function lineOf(repoId: string): Line {
  let line = lines.get(repoId);
  if (line === undefined) {
    line = { waiting: [], holds: new Set(), gating: 0, gates: 0, draining: false };
    line = { repoId, waiting: [], holds: new Set(), gating: 0, since: 0, gates: 0, draining: false };
    lines.set(repoId, line);
  }
  return line;
@@ -139,6 +175,8 @@
export interface QueueStatus {
  readonly repoId: string;
  readonly queued: number;
  readonly gating: number;
  /** When the candidate at the gate started, epoch ms; 0 when none is. */
  readonly since: number;
}
export function queueStatus(): QueueStatus[] {
@@ -147,10 +185,39 @@
export function queueStatus(): QueueStatus[] {
      repoId,
      
        junipercrew
        24d ago
        
      
      since is added to QueueStatus but no caller reads it. Surface it on the flow panel and in tty, or remove it.
    
      queued: line.waiting.length,
      gating: line.gating,
      since: line.since,
    }))
    .filter((s) => s.queued > 0 || s.gating > 0);
}
/**
 * The repository's line in a few words, for the tick line (#3430):
 * `gating 4 pulls since 18:54` (UTC), `2 waiting`, both, or `idle`.
 */
export function queueWords(repoId: string): string {
  const line = lines.get(repoId);
  if (line === undefined) return "idle";
  const waiting = line.waiting.length === 0 ? "" : `${line.waiting.length} waiting`;
  if (line.gating === 0) return waiting === "" ? "idle" : waiting;
  return `gating ${pullsWord(line.gating)} since ${hhmm(line.since)}${waiting === "" ? "" : `, ${waiting}`}`;
}
/** `1 pull`, `8 pulls`. */
function pullsWord(n: number): string {
  return n === 1 ? "1 pull" : `${n} pulls`;
}
/** An epoch-ms instant as `HH:MM`, UTC as the log is. */
function hhmm(ms: number): string {
  return new Date(ms).toISOString().slice(11, 16);
}
/** A gate's verdict in a word, for the log. */
function outcome(verdict: LandVerdict): string {
  if (verdict.lands) return "green";
  return "interrupted" in verdict ? "cut off" : "red";
}
/** Gate runs a repository's line has made since boot. */
export function gateRuns(repoId: string): number {
  return lines.get(repoId)?.gates ?? 0;
@@ -159,24 +226,57 @@
export function gateRuns(repoId: string): number {
/**
 * One candidate at a time, until the line is empty. A batch is the front
 * of the line that shares one place and one base branch; a sealed pull
 * whose objects live elsewhere is a batch of its own.
 * whose objects live elsewhere is a batch of its own. A batch carries at
 * most `MAX_CANDIDATE` pulls.
 */
async function drain(line: Line, gate: GateCandidate): Promise<void> {
  line.draining = true;
  try {
    while (line.waiting.length > 0) {
      const first = line.waiting[0]!.entry;
      const batch = line.waiting.filter(
        (w) => w.entry.dir === first.dir && w.entry.pull.baseBranch === first.pull.baseBranch,
      );
      const batch = line.waiting
        .filter((w) => w.entry.dir === first.dir && w.entry.pull.baseBranch === first.pull.baseBranch)
        .slice(0, MAX_CANDIDATE);
      line.waiting = line.waiting.filter((w) => !batch.includes(w));
      const counted: GateCandidate = async (candidate) => {
      const counted: Gated = async (candidate) => {
        line.gates++;
        line.gating = candidate.steps.length;
        line.since = Date.now();
        const pulls = candidate.steps.map((s) => `#${s.entry.pull.number}`).join(",");
        // Said as it starts and as it ends: a gate takes minutes (#3430).
        log.info(`gating ${pullsWord(line.gating)} since ${hhmm(line.since)}`, {
          repo: line.repoId,
          pulls,
          waiting: line.waiting.length,
        });
        const abandon = new AbortController();
        let timer: ReturnType<typeof setTimeout> | undefined;
        // A gate that never answers is abandoned at the deadline: no
        // verdict, and the next sweep forms the line again (#3430).
        const deadline = new Promise<LandVerdict>((resolve) => {
          timer = setTimeout(() => {
            log.warn("candidate abandoned", {
              repo: line.repoId,
              pulls,
              since: new Date(line.since).toISOString(),
              deadline: `${Math.round(deadlineMs / 1000)}s`,
            });
            abandon.abort();
            resolve({ lands: false, interrupted: true });
          }, deadlineMs);
        });
        try {
          return await gate(candidate);
          const verdict = await Promise.race([gate(candidate, abandon.signal), deadline]);
          log.info(`gated ${pullsWord(line.gating)}: ${outcome(verdict)}`, {
            repo: line.repoId,
            pulls,
            took: `${Math.round((Date.now() - line.since) / 1000)}s`,
          });
          return verdict;
        } finally {
          clearTimeout(timer);
          line.gating = 0;
          line.since = 0;
        }
      };
      let left: Waiting[];
@@ -202,7 +302,7 @@
async function drain(line: Line, gate: GateCandidate): Promise<void> {
 */
async function round(
  batch: readonly Waiting[],
  gate: GateCandidate,
  gate: Gated,
  known?: { treeOid: string; red: Red },
): Promise<Waiting[]> {
  const candidate = await build(batch);
@@ -139,11 +139,13 @@
export interface ChecksHandle {
   * queued and no run is written; the answer is the verdict. `interrupted`
   * is a gate that did not finish for a reason that is not the code — this
   * forge stopping, or the cluster losing its Job — and is no verdict (#3155).
   * `signal` aborting (the queue's deadline, #3430) cuts it off the same way.
   */
  gate(
    repo: Repo,
    pull: Pull,
    commitOid: string,
    signal?: AbortSignal,
  ): Promise<{ ok: boolean; error: string; lines: string[]; interrupted?: boolean }>;
  /**
   * Stop checking. Checks under way get `drainMs` to finish unless `drain`
@@ -1120,13 +1122,15 @@
export function startChecks(app: App, options: ChecksOptions): ChecksHandle {
  const handle: ChecksHandle = {
    request,
    gate: async (repo, pull, commitOid) => {
    gate: async (repo, pull, commitOid, signal) => {
      const lines: string[] = [];
      // A gate asked of a forge already stopping is not started: the forge
      // that carries on asks it again.
      if (stopped) return { ok: false, error: INTERRUPTED, lines, interrupted: true };
      const controller = new AbortController();
      gates.add(controller);
      const abandon = (): void => controller.abort();
      signal?.addEventListener("abort", abandon, { once: true });
      
        junipercrew
        24d ago
        
      
      If signal is already aborted when gate is called, this listener never fires. Check signal?.aborted first and abort the controller straight away.
    
      
        junipercrew
        24d ago
        
      
      The abort listener is attached after the early stopped return, which is correct. If signal is already aborted when it is passed in, addEventListener never fires. Check signal?.aborted here and abort the controller at once. The queue only passes a live signal today, so this is a guard for later callers.
    
      try {
        const bare = await pullDir(app.db, app.reposRoot, repo, pull);
        const done = await runTier({
@@ -1146,6 +1150,7 @@
export function startChecks(app: App, options: ChecksOptions): ChecksHandle {
          return { ok: false, error: done.error, lines, interrupted: true };
        return { ok: false, error: done.error, lines };
      } finally {
        signal?.removeEventListener("abort", abandon);
        gates.delete(controller);
      }
    },
@@ -23,7 +23,7 @@
import {
import type { AccessError } from "../actions/access.ts";
import { gateCandidate, releaseOnLand } from "../checks/landing.ts";
import { addLearned } from "../checks/learned.ts";
import { enqueue, type GateCandidate, type Queued } from "../checks/queue.ts";
import { enqueue, type GateCandidate, type Queued, queueWords } from "../checks/queue.ts";
import { mergeable, resolveOid, treeOf } from "../git/merge.ts";
import { repoDir } from "../git/repos.ts";
import { CLOSED_NOTE } from "../core/done_on_main.ts";
@@ -1192,6 +1192,8 @@
export async function autonomyTick(
            }),
        // The approved pulls the last green sweep saw, and why each waits (#3055).
        ...(approved === undefined ? {} : { approved }),
        // What the merge queue is doing now, not what the sweep admitted (#3430).
        queue: queueWords(current.id),
        spent: `${status.spentTodayUsd.toFixed(2)}/${status.ceilingUsd}`,
        ...(status.paused === null ? {} : { paused: status.paused }),
      });
@@ -1841,8 +1843,8 @@
export function regateDue(red: Decision, now: number): boolean {
/** The queue's gate: the full tier on the candidate's tip, asked as its first pull. */
function gateFor(app: App, repo: Repo): GateCandidate {
  return (candidate) =>
    gateCandidate(app, repo, candidate.steps[0]!.entry.pull, candidate.dir, candidate.tipOid);
  return (candidate, signal) =>
    gateCandidate(app, repo, candidate.steps[0]!.entry.pull, candidate.dir, candidate.tipOid, signal);
}
/**
@@ -13,7 +13,7 @@
import { createApp, type App } from "../src/app.ts";
import { mergePull, openPull } from "../src/actions/pulls.ts";
import { commitToBranch, createRepo } from "../src/actions/repos.ts";
import { readLearned } from "../src/checks/learned.ts";
import { enqueue, gateRuns, queueStatus, type Queued } from "../src/checks/queue.ts";
import { enqueue, gateDeadlineForTest, MAX_CANDIDATE, gateRuns, queueStatus, queueWords, type Queued } from "../src/checks/queue.ts";
import { provideChecksForTest, type ChecksHandle } from "../src/checks/service.ts";
import { latestDecision } from "../src/core/decisions.ts";
import { getPull, type Pull } from "../src/core/pulls.ts";
@@ -41,6 +41,8 @@
describe("the merge queue", () => {
  let cutOff = 0;
  /** How many gates from now go red whatever the tree, as a machine's red would. */
  let flaky = 0;
  /** The signal the last gate was handed: the queue aborts it at the deadline. */
  let lastSignal: AbortSignal | undefined;
  const git = (repo: Repo, ...args: string[]) =>
    Bun.spawnSync(["git", ...args], {
@@ -63,7 +65,8 @@
describe("the merge queue", () => {
    if (!a.ok) throw new Error(a.error);
    alice = a.value;
    provideChecksForTest(app, {
      gate: async (repo: Repo, _pull: Pull, commitOid: string) => {
      gate: async (repo: Repo, _pull: Pull, commitOid: string, signal?: AbortSignal) => {
        lastSignal = signal;
        if (held !== undefined) await held;
        if (cutOff > 0) {
          cutOff--;
@@ -180,6 +183,29 @@
describe("the merge queue", () => {
    expect(queueStatus().some((s) => s.repoId === repo.id)).toBe(false);
  });
  test("ten green pulls: a candidate carries eight, the next the other two, and the log says each gate (#3430)", async () => {
    const repo = await autonomousRepo("ten-green");
    const base = git(repo, "rev-parse", "refs/heads/main");
    const pulls = [];
    for (let i = 0; i < 10; i++) pulls.push(await greenPull(repo, `p${i}.ts`));
    const said: string[] = [];
    const info = console.info;
    console.info = (line: string) => said.push(line);
    try {
      expect(await sweepGreen(app, repo)).toBe(10);
    } finally {
      console.info = info;
    }
    expect(MAX_CANDIDATE).toBe(8);
    expect(gatesOf(repo)).toHaveLength(2);
    expect(git(repo, "rev-list", "--count", `${base}..${gatesOf(repo)[0]!.commit}`)).toBe("8");
    for (const pull of pulls) expect(merged(repo, pull)).toBe(true);
    const queue = said.filter((l) => l.includes("[queue] "));
    expect(queue.some((l) => /\[queue\] gating 8 pulls since \d\d:\d\d .*waiting=2/.test(l))).toBe(true);
    expect(queue.some((l) => l.includes("[queue] gated 8 pulls: green"))).toBe(true);
    expect(queue.some((l) => l.includes("[queue] gated 2 pulls: green"))).toBe(true);
  });
  test("one red among four: three land, the culprit is handed back, three gate runs", async () => {
    const repo = await autonomousRepo("one-red");
    const pulls = [];
@@ -363,6 +389,37 @@
describe("the merge queue", () => {
    expect(merged(repo, pull)).toBe(true);
  });
  test("a gate that never answers is abandoned at the deadline, and the next sweep lands (#3430)", async () => {
    const repo = await autonomousRepo("never-answers");
    const pull = await greenPull(repo, "a.ts");
    expect(queueWords(repo.id)).toBe("idle");
    const restore = gateDeadlineForTest(300);
    const warned: string[] = [];
    const warn = console.warn;
    console.warn = (line: string) => warned.push(line);
    held = new Promise<void>(() => {});
    try {
      lastSignal = undefined;
      const sweep = sweepGreen(app, repo);
      // Building the candidate is git work first: wait for the gate itself.
      for (let i = 0; i < 200 && lastSignal === undefined; i++) await Bun.sleep(5);
      expect(queueWords(repo.id)).toMatch(/^gating 1 pull since \d\d:\d\d$/);
      expect(await sweep).toBe(0);
    } finally {
      held = undefined;
      console.warn = warn;
      restore();
    }
    expect((lastSignal as AbortSignal | undefined)?.aborted).toBe(true);
    expect(warned.some((l) => l.includes("[queue] candidate abandoned") && l.includes(`#${pull.number}`))).toBe(true);
    expect(queueWords(repo.id)).toBe("idle");
    // No verdict: nothing told, and the line forms again on the next sweep.
    expect(latestDecision(app.db, pull.id, "land")?.answer).not.toBe("red");
    expect(listComments(app.db, pull.id)).toHaveLength(0);
    expect(await sweepGreen(app, repo)).toBe(1);
    expect(merged(repo, pull)).toBe(true);
  });
  test("a red cut off mid-bisection hands no one back; the line lands on the next sweep", async () => {
    const repo = await autonomousRepo("cut-off-bisect");
    const pulls = [];
		@@ -1087,6 +1087,10 @@ word) and `issue.parked = 1` carries it, so raw SQL never reads it as done.
1087	1087	its repository's queue (`src/checks/queue.ts`). The queue builds one
1088	1088	candidate: main, then each queued pull's squash on top of the one before, in
1089	1089	queue order, leaving out any pull that conflicts or has nothing left to land.
	1090	A candidate carries at most eight pulls (`MAX_CANDIDATE`), so a gate stays
	1091	near one full tier and a red bisects in three; the rest wait their turn.
	1092	The log says `[queue] gating N pulls since HH:MM` as a gate starts and
	1093	`[queue] gated N pulls: green\|red\|cut off` as it ends.
1090	1094	The full gate runs once on that candidate. Green lands every pull in order,
1091	1095	and each landing is refused unless main is still where the candidate was
1092	1096	built and the squash writes the tree that was gated, so main ends at the
		@@ -1096,7 +1100,9 @@ waits for a push or a new main, except once: after 30 minutes with neither
1096	1100	side moved it is gated again, and lands if green; a red that reproduces
1097	1101	stays red and is not told twice. A gate cut off by the forge stopping (a
1098	1102	roll, a drain) or by the cluster losing its Job is no red at all: the next
1099		sweep gates it again. Pulls
	1103	sweep gates it again. So is a gate that has not answered in 15 minutes
	1104	(`GATE_DEADLINE_MS`): it is abandoned and its check aborted, the log says
	1105	`[queue] candidate abandoned`, and the next sweep forms the line again. Pulls
1100	1106	that turn green while a candidate is gating wait for the next one. Each
1101	1107	queue's numbers — how many pulls are queued, how many are gating, and how
1102	1108	many landed in the last hour — show in three places: the flow panel on `/boards`, the
		@@ -1110,7 +1116,9 @@ own first; the batch lands in one gate. A pull's own check runs again only
1110	1116	when its head changes. A person's merge, which no candidate gated, still
1111	1117	refuses a stale green. The autonomy tick line counts the approved pulls the
1112	1118	last sweep saw and why each waits (`approved="3 approved: queued=2 review=1"`,
1113		from `src/crew/waits.ts`).
	1119	from `src/crew/waits.ts`), and what the queue is doing now
	1120	(`queue="gating 2 pulls since 18:54"`, `queue="1 waiting"` or `queue=idle`,
	1121	in UTC).
1114	1122
1115	1123	An autonomous app's planner is held while nothing is landing or the plate
1116	1124	is full: no pull has merged in 48 h while one was closed unmerged, or more than
		@@ -35,6 +35,7 @@ import { ownProbeHeaders } from "../kernel/probe.ts";
35	35	* pull with its base, or the merge queue's candidate of several (#2036).
36	36	* `pull` is the one the gate is asked for: where the objects are and which
37	37	* base branch's check is run. The manifest is the candidate's own.
	38	* `signal` is the queue abandoning the gate at its deadline (#3430).
38	39	*/
39	40	export async function gateCandidate(
40	41	app: App,
		@@ -42,6 +43,7 @@ export async function gateCandidate(
42	43	pull: Pull,
43	44	dir: string,
44	45	commitOid: string,
	46	signal?: AbortSignal,
45	47	): Promise<LandVerdict> {
46	48	const checks = checksFor(app);
47	49	if (checks === undefined) return { lands: true };
		@@ -52,7 +54,7 @@ export async function gateCandidate(
52	54	const declared = parseCheck(forgeJson, packageJson);
53	55	if (!declared.ok \|\| declared.value === undefined \|\| declared.value.fast === null)
54	56	return { lands: true };
55		const done = await checks.gate(repo, pull, commitOid);
	57	const done = await checks.gate(repo, pull, commitOid, signal);
56	58	// Cut off, not judged: no verdict, and not the gate log a red reads (#3155).
57	59	if (done.interrupted === true) return { lands: false, interrupted: true };
58	60	keepForgeGate(app.config.dataDir, app.config.forgeRepo, `${repo.ownerName}/${repo.name}`, done.lines);
		@@ -10,7 +10,10 @@
10	10	* search in the first half while the rest waits for the next candidate. A
11	11	* single red step is the culprit and is handed back. A gate cut off by the
12	12	* forge stopping is no verdict: nothing is handed back, and the forge that
13		* carries on admits the line again (#3155).
	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).
14	17	*
15	18	* What is queued is not stored: it is the pulls `landIfGreen` admitted, and
16	19	* every sweep admits them again. The queue only orders them and remembers
		@@ -67,8 +70,38 @@ export interface Candidate {
67	70	readonly treeOid: string;
68	71	}
69	72
70		/** Gate a candidate: the full tier on its tip. */
71		export type GateCandidate = (candidate: Candidate) => Promise<LandVerdict>;
	73	/**
	74	* Gate a candidate: the full tier on its tip. `signal` aborts when the
	75	* queue abandons the gate at its deadline: what it started should stop.
	76	*/
	77	export type GateCandidate = (candidate: Candidate, signal: AbortSignal) => Promise<LandVerdict>;
	78
	79	/**
	80	* How long a candidate's gate may run before the queue abandons it (#3430):
	81	* twice the forge's full tier. Past it the gate is no verdict, as a gate the
	82	* forge stopping cut off is not (#3155).
	83	*/
	84	export const GATE_DEADLINE_MS = 15 * 60_000;
	85
	86	let deadlineMs = GATE_DEADLINE_MS;
	87
	88	/**
	89	* The most pulls one candidate carries (#3430): eight keep a gate near one
	90	* full tier and a red's bisection three gates deep. The rest of the line
	91	* waits for the next candidate, in order.
	92	*/
	93	export const MAX_CANDIDATE = 8;
	94
	95	/** Tests shorten the deadline; the answer puts it back. */
	96	export function gateDeadlineForTest(ms: number): () => void {
	97	deadlineMs = ms;
	98	return () => {
	99	deadlineMs = GATE_DEADLINE_MS;
	100	};
	101	}
	102
	103	/** The gate as a round asks it: the deadline is the line's own. */
	104	type Gated = (candidate: Candidate) => Promise<LandVerdict>;
72	105
73	106	interface Waiting {
74	107	readonly entry: Queued;
		@@ -76,11 +109,14 @@ interface Waiting {
76	109	}
77	110
78	111	interface Line {
	112	readonly repoId: string;
79	113	waiting: Waiting[];
80	114	/** Pull ids in the queue or its candidate: a pull is in the line once. */
81	115	readonly holds: Set<string>;
82	116	/** How many pulls the candidate at the gate now carries. */
83	117	gating: number;
	118	/** When the candidate at the gate started, epoch ms; 0 when none is. */
	119	since: number;
84	120	/** The gate runs this line has made, for a test and the log. */
85	121	gates: number;
86	122	draining: boolean;
		@@ -100,7 +136,7 @@ const IDENTITY = {
100	136	function lineOf(repoId: string): Line {
101	137	let line = lines.get(repoId);
102	138	if (line === undefined) {
103		line = { waiting: [], holds: new Set(), gating: 0, gates: 0, draining: false };
	139	line = { repoId, waiting: [], holds: new Set(), gating: 0, since: 0, gates: 0, draining: false };
104	140	lines.set(repoId, line);
105	141	}
106	142	return line;
		@@ -139,6 +175,8 @@ export interface QueueStatus {
139	175	readonly repoId: string;
140	176	readonly queued: number;
141	177	readonly gating: number;
	178	/** When the candidate at the gate started, epoch ms; 0 when none is. */
	179	readonly since: number;
142	180	}
143	181
144	182	export function queueStatus(): QueueStatus[] {
		@@ -147,10 +185,39 @@ export function queueStatus(): QueueStatus[] {
147	185	repoId,
		junipercrew 24d ago `since` is added to `QueueStatus` but no caller reads it. Surface it on the flow panel and in tty, or remove it.
148	186	queued: line.waiting.length,
149	187	gating: line.gating,
	188	since: line.since,
150	189	}))
151	190	.filter((s) => s.queued > 0 \|\| s.gating > 0);
152	191	}
153	192
	193	/**
	194	* 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`.
	196	*/
	197	export function queueWords(repoId: string): string {
	198	const line = lines.get(repoId);
	199	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}`}`;
	203	}
	204
	205	/** `1 pull`, `8 pulls`. */
	206	function pullsWord(n: number): string {
	207	return n === 1 ? "1 pull" : `${n} pulls`;
	208	}
	209
	210	/** An epoch-ms instant as `HH:MM`, UTC as the log is. */
	211	function hhmm(ms: number): string {
	212	return new Date(ms).toISOString().slice(11, 16);
	213	}
	214
	215	/** A gate's verdict in a word, for the log. */
	216	function outcome(verdict: LandVerdict): string {
	217	if (verdict.lands) return "green";
	218	return "interrupted" in verdict ? "cut off" : "red";
	219	}
	220
154	221	/** Gate runs a repository's line has made since boot. */
155	222	export function gateRuns(repoId: string): number {
156	223	return lines.get(repoId)?.gates ?? 0;
		@@ -159,24 +226,57 @@ export function gateRuns(repoId: string): number {
159	226	/**
160	227	* One candidate at a time, until the line is empty. A batch is the front
161	228	* of the line that shares one place and one base branch; a sealed pull
162		* whose objects live elsewhere is a batch of its own.
	229	* whose objects live elsewhere is a batch of its own. A batch carries at
	230	* most `MAX_CANDIDATE` pulls.
163	231	*/
164	232	async function drain(line: Line, gate: GateCandidate): Promise<void> {
165	233	line.draining = true;
166	234	try {
167	235	while (line.waiting.length > 0) {
168	236	const first = line.waiting[0]!.entry;
169		const batch = line.waiting.filter(
170		(w) => w.entry.dir === first.dir && w.entry.pull.baseBranch === first.pull.baseBranch,
171		);
	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);
172	240	line.waiting = line.waiting.filter((w) => !batch.includes(w));
173		const counted: GateCandidate = async (candidate) => {
	241	const counted: Gated = async (candidate) => {
174	242	line.gates++;
175	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)}`, {
	248	repo: line.repoId,
	249	pulls,
	250	waiting: line.waiting.length,
	251	});
	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	});
176	268	try {
177		return await gate(candidate);
	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;
178	276	} finally {
	277	clearTimeout(timer);
179	278	line.gating = 0;
	279	line.since = 0;
180	280	}
181	281	};
182	282	let left: Waiting[];
		@@ -202,7 +302,7 @@ async function drain(line: Line, gate: GateCandidate): Promise<void> {
202	302	*/
203	303	async function round(
204	304	batch: readonly Waiting[],
205		gate: GateCandidate,
	305	gate: Gated,
206	306	known?: { treeOid: string; red: Red },
207	307	): Promise<Waiting[]> {
208	308	const candidate = await build(batch);
		@@ -139,11 +139,13 @@ export interface ChecksHandle {
139	139	* queued and no run is written; the answer is the verdict. `interrupted`
140	140	* is a gate that did not finish for a reason that is not the code — this
141	141	* forge stopping, or the cluster losing its Job — and is no verdict (#3155).
	142	* `signal` aborting (the queue's deadline, #3430) cuts it off the same way.
142	143	*/
143	144	gate(
144	145	repo: Repo,
145	146	pull: Pull,
146	147	commitOid: string,
	148	signal?: AbortSignal,
147	149	): Promise<{ ok: boolean; error: string; lines: string[]; interrupted?: boolean }>;
148	150	/**
149	151	* Stop checking. Checks under way get `drainMs` to finish unless `drain`
		@@ -1120,13 +1122,15 @@ export function startChecks(app: App, options: ChecksOptions): ChecksHandle {
1120	1122
1121	1123	const handle: ChecksHandle = {
1122	1124	request,
1123		gate: async (repo, pull, commitOid) => {
	1125	gate: async (repo, pull, commitOid, signal) => {
1124	1126	const lines: string[] = [];
1125	1127	// A gate asked of a forge already stopping is not started: the forge
1126	1128	// that carries on asks it again.
1127	1129	if (stopped) return { ok: false, error: INTERRUPTED, lines, interrupted: true };
1128	1130	const controller = new AbortController();
1129	1131	gates.add(controller);
	1132	const abandon = (): void => controller.abort();
	1133	signal?.addEventListener("abort", abandon, { once: true });
		junipercrew 24d ago If `signal` is already aborted when `gate` is called, this listener never fires. Check `signal?.aborted` first and abort the controller straight away. junipercrew 24d ago The abort listener is attached after the early `stopped` return, which is correct. If `signal` is already aborted when it is passed in, `addEventListener` never fires. Check `signal?.aborted` here and abort the controller at once. The queue only passes a live signal today, so this is a guard for later callers.
1130	1134	try {
1131	1135	const bare = await pullDir(app.db, app.reposRoot, repo, pull);
1132	1136	const done = await runTier({
		@@ -1146,6 +1150,7 @@ export function startChecks(app: App, options: ChecksOptions): ChecksHandle {
1146	1150	return { ok: false, error: done.error, lines, interrupted: true };
1147	1151	return { ok: false, error: done.error, lines };
1148	1152	} finally {
	1153	signal?.removeEventListener("abort", abandon);
1149	1154	gates.delete(controller);
1150	1155	}
1151	1156	},
		@@ -23,7 +23,7 @@ import {
23	23	import type { AccessError } from "../actions/access.ts";
24	24	import { gateCandidate, releaseOnLand } from "../checks/landing.ts";
25	25	import { addLearned } from "../checks/learned.ts";
26		import { enqueue, type GateCandidate, type Queued } from "../checks/queue.ts";
	26	import { enqueue, type GateCandidate, type Queued, queueWords } from "../checks/queue.ts";
27	27	import { mergeable, resolveOid, treeOf } from "../git/merge.ts";
28	28	import { repoDir } from "../git/repos.ts";
29	29	import { CLOSED_NOTE } from "../core/done_on_main.ts";
		@@ -1192,6 +1192,8 @@ export async function autonomyTick(
1192	1192	}),
1193	1193	// The approved pulls the last green sweep saw, and why each waits (#3055).
1194	1194	...(approved === undefined ? {} : { approved }),
	1195	// What the merge queue is doing now, not what the sweep admitted (#3430).
	1196	queue: queueWords(current.id),
1195	1197	spent: `${status.spentTodayUsd.toFixed(2)}/${status.ceilingUsd}`,
1196	1198	...(status.paused === null ? {} : { paused: status.paused }),
1197	1199	});
		@@ -1841,8 +1843,8 @@ export function regateDue(red: Decision, now: number): boolean {
1841	1843
1842	1844	/** The queue's gate: the full tier on the candidate's tip, asked as its first pull. */
1843	1845	function gateFor(app: App, repo: Repo): GateCandidate {
1844		return (candidate) =>
1845		gateCandidate(app, repo, candidate.steps[0]!.entry.pull, candidate.dir, candidate.tipOid);
	1846	return (candidate, signal) =>
	1847	gateCandidate(app, repo, candidate.steps[0]!.entry.pull, candidate.dir, candidate.tipOid, signal);
1846	1848	}
1847	1849
1848	1850	/**
		@@ -13,7 +13,7 @@ import { createApp, type App } from "../src/app.ts";
13	13	import { mergePull, openPull } from "../src/actions/pulls.ts";
14	14	import { commitToBranch, createRepo } from "../src/actions/repos.ts";
15	15	import { readLearned } from "../src/checks/learned.ts";
16		import { enqueue, gateRuns, queueStatus, type Queued } from "../src/checks/queue.ts";
	16	import { enqueue, gateDeadlineForTest, MAX_CANDIDATE, gateRuns, queueStatus, queueWords, type Queued } from "../src/checks/queue.ts";
17	17	import { provideChecksForTest, type ChecksHandle } from "../src/checks/service.ts";
18	18	import { latestDecision } from "../src/core/decisions.ts";
19	19	import { getPull, type Pull } from "../src/core/pulls.ts";
		@@ -41,6 +41,8 @@ describe("the merge queue", () => {
41	41	let cutOff = 0;
42	42	/** How many gates from now go red whatever the tree, as a machine's red would. */
43	43	let flaky = 0;
	44	/** The signal the last gate was handed: the queue aborts it at the deadline. */
	45	let lastSignal: AbortSignal \| undefined;
44	46
45	47	const git = (repo: Repo, ...args: string[]) =>
46	48	Bun.spawnSync(["git", ...args], {
		@@ -63,7 +65,8 @@ describe("the merge queue", () => {
63	65	if (!a.ok) throw new Error(a.error);
64	66	alice = a.value;
65	67	provideChecksForTest(app, {
66		gate: async (repo: Repo, _pull: Pull, commitOid: string) => {
	68	gate: async (repo: Repo, _pull: Pull, commitOid: string, signal?: AbortSignal) => {
	69	lastSignal = signal;
67	70	if (held !== undefined) await held;
68	71	if (cutOff > 0) {
69	72	cutOff--;
		@@ -180,6 +183,29 @@ describe("the merge queue", () => {
180	183	expect(queueStatus().some((s) => s.repoId === repo.id)).toBe(false);
181	184	});
182	185
	186	test("ten green pulls: a candidate carries eight, the next the other two, and the log says each gate (#3430)", async () => {
	187	const repo = await autonomousRepo("ten-green");
	188	const base = git(repo, "rev-parse", "refs/heads/main");
	189	const pulls = [];
	190	for (let i = 0; i < 10; i++) pulls.push(await greenPull(repo, `p${i}.ts`));
	191	const said: string[] = [];
	192	const info = console.info;
	193	console.info = (line: string) => said.push(line);
	194	try {
	195	expect(await sweepGreen(app, repo)).toBe(10);
	196	} finally {
	197	console.info = info;
	198	}
	199	expect(MAX_CANDIDATE).toBe(8);
	200	expect(gatesOf(repo)).toHaveLength(2);
	201	expect(git(repo, "rev-list", "--count", `${base}..${gatesOf(repo)[0]!.commit}`)).toBe("8");
	202	for (const pull of pulls) expect(merged(repo, pull)).toBe(true);
	203	const queue = said.filter((l) => l.includes("[queue] "));
	204	expect(queue.some((l) => /\[queue\] gating 8 pulls since \d\d:\d\d .*waiting=2/.test(l))).toBe(true);
	205	expect(queue.some((l) => l.includes("[queue] gated 8 pulls: green"))).toBe(true);
	206	expect(queue.some((l) => l.includes("[queue] gated 2 pulls: green"))).toBe(true);
	207	});
	208
183	209	test("one red among four: three land, the culprit is handed back, three gate runs", async () => {
184	210	const repo = await autonomousRepo("one-red");
185	211	const pulls = [];
		@@ -363,6 +389,37 @@ describe("the merge queue", () => {
363	389	expect(merged(repo, pull)).toBe(true);
364	390	});
365	391
	392	test("a gate that never answers is abandoned at the deadline, and the next sweep lands (#3430)", async () => {
	393	const repo = await autonomousRepo("never-answers");
	394	const pull = await greenPull(repo, "a.ts");
	395	expect(queueWords(repo.id)).toBe("idle");
	396	const restore = gateDeadlineForTest(300);
	397	const warned: string[] = [];
	398	const warn = console.warn;
	399	console.warn = (line: string) => warned.push(line);
	400	held = new Promise<void>(() => {});
	401	try {
	402	lastSignal = undefined;
	403	const sweep = sweepGreen(app, repo);
	404	// Building the candidate is git work first: wait for the gate itself.
	405	for (let i = 0; i < 200 && lastSignal === undefined; i++) await Bun.sleep(5);
	406	expect(queueWords(repo.id)).toMatch(/^gating 1 pull since \d\d:\d\d$/);
	407	expect(await sweep).toBe(0);
	408	} finally {
	409	held = undefined;
	410	console.warn = warn;
	411	restore();
	412	}
	413	expect((lastSignal as AbortSignal \| undefined)?.aborted).toBe(true);
	414	expect(warned.some((l) => l.includes("[queue] candidate abandoned") && l.includes(`#${pull.number}`))).toBe(true);
	415	expect(queueWords(repo.id)).toBe("idle");
	416	// No verdict: nothing told, and the line forms again on the next sweep.
	417	expect(latestDecision(app.db, pull.id, "land")?.answer).not.toBe("red");
	418	expect(listComments(app.db, pull.id)).toHaveLength(0);
	419	expect(await sweepGreen(app, repo)).toBe(1);
	420	expect(merged(repo, pull)).toBe(true);
	421	});
	422
366	423	test("a red cut off mid-bisection hands no one back; the line lands on the next sweep", async () => {
367	424	const repo = await autonomousRepo("cut-off-bisect");
368	425	const pulls = [];