4da75bb · 53c72ff · 4dc4731 · 0e8d93c
@@ -208,6 +208,8 @@
interface Line {
  readonly holds: Set<string>;
  /** The gates this line has in flight now. */
  readonly flights: Set<InFlight>;
  /** Candidates being built, taken from the line and not yet at the gate. */
  forming: number;
  /** The gate runs this line has made, for a test and the log. */
  gates: number;
  /** Pulls handed back red, by pull id: re-checked alone if they join again. */
@@ -249,6 +251,44 @@
export function extraGates(): number {
  return extra;
}
const watchers = new Set<() => void>();
/**
 * Calls `fn` whenever a gate's slot frees: a gate ends or a candidate stops
 * forming. The check service schedules then, not at its next tick, so a pull
 * check held back for the gate starts as soon as the gate is done.
 */
export function watchGates(fn: () => void): () => void {
      
        sentinelcrew
        19d ago
        outdated
      
      line.flights.size === 0 makes this false the moment the gate starts, and that is when the gate needs the slot. As written, it covers only the seconds a git merge takes.
    
  watchers.add(fn);
  return () => void watchers.delete(fn);
}
function gateFreed(): void {
  for (const fn of [...watchers]) fn();
}
/**
 * Every gate in flight, each line's first included: each holds a check slot
 * for as long as it runs, so a pull check never shares the machine with it
 * (`checks/service.ts`, #6376).
 */
export function gatesInFlight(): number {
  let n = extra;
  for (const line of lines.values()) if (line.flights.size > 0) n++;
  return n;
}
/**
 * Whether some line's first gate is about to start: its candidate is being
 * built and no gate of that line is in flight. Such a gate takes the next
 * free check slot before a pull check does (`checks/service.ts`); once it
 * runs, `gatesInFlight` holds the slot.
 */
export function gateQueued(): boolean {
  for (const line of lines.values()) if (line.forming > 0 && line.flights.size === 0) return true;
  return false;
}
const IDENTITY = {
  GIT_AUTHOR_NAME: "forge",
  GIT_AUTHOR_EMAIL: "forge@localhost",
@@ -266,6 +306,7 @@
function lineOf(repoId: string): Line {
      waiting: [],
      holds: new Set(),
      flights: new Set(),
      forming: 0,
      gates: 0,
      reds: new Map(),
      suspects: new Map(),
@@ -464,6 +505,7 @@
function counted(line: Line, gate: GateCandidate): Gated {
      if (dropped !== undefined) signal?.removeEventListener("abort", dropped);
      line.flights.delete(flight);
      if (beyond) extra--;
      gateFreed();
    }
  };
}
@@ -643,21 +685,29 @@
async function launch(line: Line, gated: Gated, on: Flight | undefined): Promise
    : [...same.filter((w) => !cleared(w)), ...same.filter(cleared)].slice(0, MAX_CANDIDATE);
  line.waiting = line.waiting.filter((w) => !batch.includes(w));
  let candidate: Candidate | undefined;
  try {
    candidate = await build(batch, on?.candidate);
  } catch (error) {
    log.warn("merge queue round failed", { error: String(error) });
  }
  if (candidate === undefined || candidate.steps.length === 0) {
    for (const w of batch) w.settle(false);
    return undefined;
  }
  const steps = inCandidate(batch, candidate);
  line.forming++;
  let verdict: Promise<LandVerdict>;
  let steps: Waiting[];
  const abort = new AbortController();
  const verdict = gated(candidate, abort.signal).catch((error: unknown) => {
    log.warn("merge queue round failed", { error: String(error) });
    return UNDECIDED;
  });
  try {
    try {
      candidate = await build(batch, on?.candidate);
    } catch (error) {
      log.warn("merge queue round failed", { error: String(error) });
    }
    if (candidate === undefined || candidate.steps.length === 0) {
      for (const w of batch) w.settle(false);
      return undefined;
    }
    steps = inCandidate(batch, candidate);
    verdict = gated(candidate, abort.signal).catch((error: unknown) => {
      log.warn("merge queue round failed", { error: String(error) });
      return UNDECIDED;
    });
  } finally {
    line.forming--;
    gateFreed();
  }
  return { steps, candidate, speculative: on !== undefined, recheck: front.recheck, branch, verdict, abort };
}
@@ -89,7 +89,7 @@
import type { Result } from "../kernel/result.ts";
import { placementFromEnv, type Placement } from "./placement.ts";
import { JOB_VANISHED, runJobStep } from "./job_step.ts";
import { checkInstallCache, installWithRetry, withLock } from "./install.ts";
import { extraGates } from "./queue.ts";
import { extraGates, gateQueued, gatesInFlight, watchGates } from "./queue.ts";
import type { JobOptions } from "../cluster/jobs.ts";
import type { CheckDetail } from "../core/runs.ts";
@@ -204,6 +204,15 @@
const TICK_MS = 5_000;
 */
export const STARVATION_MS = BUDGETS.laneWait / 2;
/**
 * Whether a pull's check must pass over the free slots left (`free`): while
 * a merge-queue gate is about to start (`gateQueued`), the last one is the
 * gate's.
 */
export function holdsLastSlot(free: number, gateWaits: boolean): boolean {
  return gateWaits && free <= 1;
}
/**
 * Which queued check takes a slot when one frees: the one requested most
 * recently, so a push's own check is never stuck behind pulls that queued
@@ -451,8 +460,8 @@
export function startChecks(app: App, options: ChecksOptions): ChecksHandle {
    if (stopped) return;
    const limit = capacity();
    if (options.gate === undefined) gate.resize(limit);
    // A merge-queue gate beyond a line's first holds a slot (#4053).
    while (inFlight.size < limit - extraGates()) {
    // Every merge-queue gate in flight holds a slot, a line's first too (#6376).
    while (inFlight.size < limit - gatesInFlight()) {
      // A pull's check waits while its main is red (`checks/red_main.ts`).
      const queued = unheld(
        listActive(app.db, "check").filter(
@@ -469,7 +478,10 @@
export function startChecks(app: App, options: ChecksOptions): ChecksHandle {
      // burst of one account's checks cannot pass together. A rebase's
      // check goes before any other (`finishing_first.ts`, #6140).
      const { rebases, rest: others } = rebasesFirst(app.db, queued);
      const next = startable(rebases) ?? startable(others);
      // While a gate is about to start, the last free slot is its: a pull's
      // check waits for it, a rebase's and a main check do not.
      const held = holdsLastSlot(limit - gatesInFlight() - inFlight.size, gateQueued());
      
        sentinelcrew
        19d ago
        outdated
      
      limit - extraGates() - inFlight.size leaves out the line's first gate once it is running, so this hold ends when the build ends and a pull check then takes the gate's slot for the whole gate. Count the first in-flight gate here, or hold while it runs.
    
      const next = startable(rebases) ?? startable(held ? others.filter((run) => run.issueId === null) : others);
      if (!next) return;
      const repo = getRepoById(app.db, next.repoId);
      if (!repo) return;
@@ -1199,6 +1211,8 @@
export function startChecks(app: App, options: ChecksOptions): ChecksHandle {
  const timer = setInterval(() => {
    if (!stopped) schedule();
  }, TICK_MS);
  // A gate's slot frees: the pull check held for it starts now, not at the tick.
  const unwatch = watchGates(schedule);
  schedule();
  void catchUp();
@@ -1248,6 +1262,7 @@
export function startChecks(app: App, options: ChecksOptions): ChecksHandle {
      stopped = true;
      clearInterval(timer);
      unsubscribe();
      unwatch();
      // A roll waits for the checks under way, up to the budget, and only
      // then cuts off what is left (it is queued again, not lost).
      const running = Promise.all([...inFlight.values()].map((f) => f.promise));
@@ -18,6 +18,7 @@
import { summariseCheckFailure } from "../src/checks/summary.ts";
import { SLOW_SECONDS, summaryLines } from "../scripts/gate-summary.ts";
import { tally } from "../scripts/shards.ts";
import {
  holdsLastSlot,
  pickNext,
  requeueAbandonedChecks,
  startChecks,
@@ -832,6 +833,15 @@
describe("checks", () => {
      ).toBe(true);
    }, 60_000);
  test("a queued gate keeps the last free slot from pull checks", () => {
    // One slot free, a gate queued: the pull check passes, the gate starts.
    expect(holdsLastSlot(1, true)).toBe(true);
    // With no gate queued the pull check takes it, as every slot.
    expect(holdsLastSlot(1, false)).toBe(false);
    // More than one free: the pull check starts, one is still left.
    expect(holdsLastSlot(2, true)).toBe(false);
  });
  test("pickNext takes the newest queued check, unless one has starved", () => {
    const a = createRun(app.db, { repoId: repo.id, kind: "check", env: "a" });
    const b = createRun(app.db, { repoId: repo.id, kind: "check", env: "b" });
@@ -0,0 +1,193 @@
/**
 * The last free check slot belongs to a merge-queue gate (#6376), from the
 * time its candidate forms until its verdict: a pull's check waits for it,
 * a rebase's check and main's check do not, and with no gate every slot is
 * the checks'. The gates are real `enqueue` calls; the service schedules
 * once as it starts, and the tests wait on a run's state or a gate's own
 * promise, never on a fixed delay.
 */
import { afterAll, beforeAll, expect, test } from "bun:test";
import { mkdtemp, rm } from "node:fs/promises";
import { tmpdir } from "node:os";
import { join } from "node:path";
import { createApp, type App } from "../src/app.ts";
import { commitToBranch, createRepo } from "../src/actions/repos.ts";
import { openPull } from "../src/actions/pulls.ts";
import { type Candidate, enqueue, gateQueued, gatesInFlight, type Queued } from "../src/checks/queue.ts";
import { startChecks } from "../src/checks/service.ts";
import type { Pull } from "../src/core/pulls.ts";
import type { Repo } from "../src/core/repos.ts";
import { cancelRun, createRun, getRun, type Run } from "../src/core/runs.ts";
import { createUser, type User } from "../src/core/users.ts";
import { serve, type Servers } from "../src/http/server.ts";
let dataDir: string;
let app: App;
let servers: Servers;
let alice: User;
let repo: Repo;
let pulls = 0;
beforeAll(async () => {
  dataDir = await mkdtemp(join(tmpdir(), "forge-gate-last-slot-"));
  app = createApp({ dataDir, port: 0 });
  servers = await serve(app);
  const a = await createUser(app.db, {
    origin: "seed",
    trusted: true,
    username: "alice",
    email: "[email protected]",
    password: "a-good-passphrase",
    adminIfFirst: true,
  });
  if (!a.ok) throw new Error(a.error);
  alice = a.value;
  const made = await createRepo(app, alice, { name: "lib", visibility: "public" });
  if (!made.ok) throw new Error(String(made.error));
  repo = made.value;
  const seeded = await commitToBranch(app, alice, repo, {
    message: "Start",
    files: [{ path: "README.md", content: "# lib\n" }],
  });
  if (!seeded.ok) throw new Error(JSON.stringify(seeded.error));
});
afterAll(async () => {
  servers.stop();
  app.close();
  await rm(dataDir, { recursive: true, force: true });
});
async function newPull(): Promise<Pull> {
  const branch = `b${++pulls}`;
  const branched = await commitToBranch(app, alice, repo, {
    branch,
    message: `Add ${branch}`,
    files: [{ path: `${branch}.md`, content: "ok\n" }],
  });
  if (!branched.ok) throw new Error(JSON.stringify(branched.error));
  const opened = await openPull(app, alice, repo, { title: branch, baseBranch: "main", headBranch: branch });
  if (!opened.ok) throw new Error(String(opened.error));
  return opened.value;
}
const pullCheck = (pull: Pull): Run =>
  createRun(app.db, { repoId: repo.id, kind: "check", env: `pull-${pull.number}`, issueId: pull.id, commitOid: pull.headOid });
/** A gate whose candidate cannot be built: it is forming now, and over in a moment. */
function formingGate(pull: Pull): Promise<boolean[]> {
  const entry: Queued = {
    pull,
    dir: join(dataDir, "nowhere"),
    land: async () => undefined,
    release: () => {},
    handBack: () => {},
  };
  const answer = enqueue(`gate-${pull.id}`, [entry], async () => ({ lands: false, interrupted: true }));
  expect(gateQueued()).toBe(true);
  return answer;
}
/** Poll until `ok`, giving up at a generous deadline rather than asserting a speed. */
async function until(ok: () => boolean): Promise<boolean> {
  for (let i = 0; i < 2000 && !ok(); i++) await Bun.sleep(5);
  return ok();
}
      
        sentinelcrew
        19d ago
        outdated
      
      This is a wall-clock wait. The "not queued" cases pass only if the service starts the run within 300 ms, which can flake on a loaded check machine. Wait on the state change instead. Also add the case that matters: a gate that is still running keeps the slot.
    
const state = (run: Run): Run["state"] => getRun(app.db, run.id)!.state;
/** Whether `run` leaves the queue once the service, with one slot, has had its turn. */
async function starts(run: Run, gate: Promise<boolean[]> | undefined): Promise<boolean> {
  const checks = startChecks(app, { forgeUrl: servers.url, concurrency: 1 });
  try {
    await gate;
    return await until(() => state(run) !== "queued");
  } finally {
    await checks.stop({ drain: false });
    cancelRun(app.db, run.id);
  }
}
test("a pull's check does not take the last slot while a gate is forming", async () => {
  const run = pullCheck(await newPull());
  const gate = formingGate(await newPull());
  const checks = startChecks(app, { forgeUrl: servers.url, concurrency: 1 });
  try {
    // The service scheduled as it started, with the gate's candidate forming.
    expect(state(run)).toBe("queued");
    await gate;
    expect(await until(() => state(run) !== "queued")).toBe(true);
  } finally {
    await checks.stop({ drain: false });
    cancelRun(app.db, run.id);
  }
  expect(gateQueued()).toBe(false);
});
test("a gate that is running keeps the slot: the pull's check starts when it settles", async () => {
  const dir = await mkdtemp(join(tmpdir(), "forge-gate-last-slot-git-"));
  const git = (...args: string[]) =>
    Bun.spawnSync(["git", "-C", dir, "-c", "user.name=t", "-c", "user.email=t@t", ...args]).stdout.toString().trim();
  Bun.spawnSync(["git", "init", "-q", "-b", "main", dir]);
  git("commit", "-q", "--allow-empty", "-m", "start");
  git("checkout", "-q", "-b", "feature");
  await Bun.write(join(dir, "f.txt"), "f\n");
  git("add", "f.txt");
  git("commit", "-q", "-m", "f");
  const headOid = git("rev-parse", "HEAD");
  git("checkout", "-q", "--detach", "main");
  const gated = Promise.withResolvers<void>();
  let running = false;
  const entry: Queued = {
    pull: { id: "gate-running", number: 9999, baseBranch: "main", headOid } as Pull,
    dir,
    land: async () => undefined,
    release: () => {},
    handBack: () => {},
  };
  const answer = enqueue("gate-running-line", [entry], async (_candidate: Candidate) => {
    running = true;
    await gated.promise;
    return { lands: false, interrupted: true };
  });
  const run = pullCheck(await newPull());
  let checks: ReturnType<typeof startChecks> | undefined;
  try {
    expect(await until(() => running)).toBe(true);
    // The candidate is built and the gate runs: nothing is forming, a gate is in flight.
    expect(gateQueued()).toBe(false);
    expect(gatesInFlight()).toBe(1);
    checks = startChecks(app, { forgeUrl: servers.url, concurrency: 1 });
    // The service scheduled as it started, and the one slot is the gate's.
    expect(state(run)).toBe("queued");
    await Bun.sleep(0);
    expect(state(run)).toBe("queued");
    gated.resolve();
    await answer;
    expect(await until(() => state(run) !== "queued")).toBe(true);
  } finally {
    gated.resolve();
    await answer;
    await checks?.stop({ drain: false });
    cancelRun(app.db, run.id);
    await rm(dir, { recursive: true, force: true });
  }
  expect(gatesInFlight()).toBe(0);
});
test("with no gate forming, the pull's check takes the slot", async () => {
  const run = pullCheck(await newPull());
  expect(gateQueued()).toBe(false);
  expect(await starts(run, undefined)).toBe(true);
});
test("a rebase's check and main's check still start while a gate is forming", async () => {
  const pull = await newPull();
  createRun(app.db, { repoId: repo.id, kind: "agent", issueId: pull.id, reason: "conflict" });
  const rebase = pullCheck(pull);
  expect(await starts(rebase, formingGate(await newPull()))).toBe(true);
  const main = createRun(app.db, { repoId: repo.id, kind: "check", env: "main", commitOid: pull.headOid });
  expect(await starts(main, formingGate(await newPull()))).toBe(true);
});
		@@ -208,6 +208,8 @@ interface Line {
208	208	readonly holds: Set<string>;
209	209	/** The gates this line has in flight now. */
210	210	readonly flights: Set<InFlight>;
	211	/** Candidates being built, taken from the line and not yet at the gate. */
	212	forming: number;
211	213	/** The gate runs this line has made, for a test and the log. */
212	214	gates: number;
213	215	/** Pulls handed back red, by pull id: re-checked alone if they join again. */
		@@ -249,6 +251,44 @@ export function extraGates(): number {
249	251	return extra;
250	252	}
251	253
	254	const watchers = new Set<() => void>();
	255
	256	/**
	257	* Calls `fn` whenever a gate's slot frees: a gate ends or a candidate stops
	258	* forming. The check service schedules then, not at its next tick, so a pull
	259	* check held back for the gate starts as soon as the gate is done.
	260	*/
	261	export function watchGates(fn: () => void): () => void {
		sentinelcrew 19d ago outdated `line.flights.size === 0` makes this false the moment the gate starts, and that is when the gate needs the slot. As written, it covers only the seconds a `git merge` takes.
	262	watchers.add(fn);
	263	return () => void watchers.delete(fn);
	264	}
	265
	266	function gateFreed(): void {
	267	for (const fn of [...watchers]) fn();
	268	}
	269
	270	/**
	271	* Every gate in flight, each line's first included: each holds a check slot
	272	* for as long as it runs, so a pull check never shares the machine with it
	273	* (`checks/service.ts`, #6376).
	274	*/
	275	export function gatesInFlight(): number {
	276	let n = extra;
	277	for (const line of lines.values()) if (line.flights.size > 0) n++;
	278	return n;
	279	}
	280
	281	/**
	282	* Whether some line's first gate is about to start: its candidate is being
	283	* built and no gate of that line is in flight. Such a gate takes the next
	284	* free check slot before a pull check does (`checks/service.ts`); once it
	285	* runs, `gatesInFlight` holds the slot.
	286	*/
	287	export function gateQueued(): boolean {
	288	for (const line of lines.values()) if (line.forming > 0 && line.flights.size === 0) return true;
	289	return false;
	290	}
	291
252	292	const IDENTITY = {
253	293	GIT_AUTHOR_NAME: "forge",
254	294	GIT_AUTHOR_EMAIL: "forge@localhost",
		@@ -266,6 +306,7 @@ function lineOf(repoId: string): Line {
266	306	waiting: [],
267	307	holds: new Set(),
268	308	flights: new Set(),
	309	forming: 0,
269	310	gates: 0,
270	311	reds: new Map(),
271	312	suspects: new Map(),
		@@ -464,6 +505,7 @@ function counted(line: Line, gate: GateCandidate): Gated {
464	505	if (dropped !== undefined) signal?.removeEventListener("abort", dropped);
465	506	line.flights.delete(flight);
466	507	if (beyond) extra--;
	508	gateFreed();
467	509	}
468	510	};
469	511	}
		@@ -643,21 +685,29 @@ async function launch(line: Line, gated: Gated, on: Flight \| undefined): Promise
643	685	: [...same.filter((w) => !cleared(w)), ...same.filter(cleared)].slice(0, MAX_CANDIDATE);
644	686	line.waiting = line.waiting.filter((w) => !batch.includes(w));
645	687	let candidate: Candidate \| undefined;
646		try {
647		candidate = await build(batch, on?.candidate);
648		} catch (error) {
649		log.warn("merge queue round failed", { error: String(error) });
650		}
651		if (candidate === undefined \|\| candidate.steps.length === 0) {
652		for (const w of batch) w.settle(false);
653		return undefined;
654		}
655		const steps = inCandidate(batch, candidate);
	688	line.forming++;
	689	let verdict: Promise<LandVerdict>;
	690	let steps: Waiting[];
656	691	const abort = new AbortController();
657		const verdict = gated(candidate, abort.signal).catch((error: unknown) => {
658		log.warn("merge queue round failed", { error: String(error) });
659		return UNDECIDED;
660		});
	692	try {
	693	try {
	694	candidate = await build(batch, on?.candidate);
	695	} catch (error) {
	696	log.warn("merge queue round failed", { error: String(error) });
	697	}
	698	if (candidate === undefined \|\| candidate.steps.length === 0) {
	699	for (const w of batch) w.settle(false);
	700	return undefined;
	701	}
	702	steps = inCandidate(batch, candidate);
	703	verdict = gated(candidate, abort.signal).catch((error: unknown) => {
	704	log.warn("merge queue round failed", { error: String(error) });
	705	return UNDECIDED;
	706	});
	707	} finally {
	708	line.forming--;
	709	gateFreed();
	710	}
661	711	return { steps, candidate, speculative: on !== undefined, recheck: front.recheck, branch, verdict, abort };
662	712	}
663	713
		@@ -89,7 +89,7 @@ import type { Result } from "../kernel/result.ts";
89	89	import { placementFromEnv, type Placement } from "./placement.ts";
90	90	import { JOB_VANISHED, runJobStep } from "./job_step.ts";
91	91	import { checkInstallCache, installWithRetry, withLock } from "./install.ts";
92		import { extraGates } from "./queue.ts";
	92	import { extraGates, gateQueued, gatesInFlight, watchGates } from "./queue.ts";
93	93	import type { JobOptions } from "../cluster/jobs.ts";
94	94	import type { CheckDetail } from "../core/runs.ts";
95	95
		@@ -204,6 +204,15 @@ const TICK_MS = 5_000;
204	204	*/
205	205	export const STARVATION_MS = BUDGETS.laneWait / 2;
206	206
	207	/**
	208	* Whether a pull's check must pass over the free slots left (`free`): while
	209	* a merge-queue gate is about to start (`gateQueued`), the last one is the
	210	* gate's.
	211	*/
	212	export function holdsLastSlot(free: number, gateWaits: boolean): boolean {
	213	return gateWaits && free <= 1;
	214	}
	215
207	216	/**
208	217	* Which queued check takes a slot when one frees: the one requested most
209	218	* recently, so a push's own check is never stuck behind pulls that queued
		@@ -451,8 +460,8 @@ export function startChecks(app: App, options: ChecksOptions): ChecksHandle {
451	460	if (stopped) return;
452	461	const limit = capacity();
453	462	if (options.gate === undefined) gate.resize(limit);
454		// A merge-queue gate beyond a line's first holds a slot (#4053).
455		while (inFlight.size < limit - extraGates()) {
	463	// Every merge-queue gate in flight holds a slot, a line's first too (#6376).
	464	while (inFlight.size < limit - gatesInFlight()) {
456	465	// A pull's check waits while its main is red (`checks/red_main.ts`).
457	466	const queued = unheld(
458	467	listActive(app.db, "check").filter(
		@@ -469,7 +478,10 @@ export function startChecks(app: App, options: ChecksOptions): ChecksHandle {
469	478	// burst of one account's checks cannot pass together. A rebase's
470	479	// check goes before any other (`finishing_first.ts`, #6140).
471	480	const { rebases, rest: others } = rebasesFirst(app.db, queued);
472		const next = startable(rebases) ?? startable(others);
	481	// While a gate is about to start, the last free slot is its: a pull's
	482	// check waits for it, a rebase's and a main check do not.
	483	const held = holdsLastSlot(limit - gatesInFlight() - inFlight.size, gateQueued());
		sentinelcrew 19d ago outdated `limit - extraGates() - inFlight.size` leaves out the line's first gate once it is running, so this hold ends when the build ends and a pull check then takes the gate's slot for the whole gate. Count the first in-flight gate here, or hold while it runs.
	484	const next = startable(rebases) ?? startable(held ? others.filter((run) => run.issueId === null) : others);
473	485	if (!next) return;
474	486	const repo = getRepoById(app.db, next.repoId);
475	487	if (!repo) return;
		@@ -1199,6 +1211,8 @@ export function startChecks(app: App, options: ChecksOptions): ChecksHandle {
1199	1211	const timer = setInterval(() => {
1200	1212	if (!stopped) schedule();
1201	1213	}, TICK_MS);
	1214	// A gate's slot frees: the pull check held for it starts now, not at the tick.
	1215	const unwatch = watchGates(schedule);
1202	1216	schedule();
1203	1217	void catchUp();
1204	1218
		@@ -1248,6 +1262,7 @@ export function startChecks(app: App, options: ChecksOptions): ChecksHandle {
1248	1262	stopped = true;
1249	1263	clearInterval(timer);
1250	1264	unsubscribe();
	1265	unwatch();
1251	1266	// A roll waits for the checks under way, up to the budget, and only
1252	1267	// then cuts off what is left (it is queued again, not lost).
1253	1268	const running = Promise.all([...inFlight.values()].map((f) => f.promise));
		@@ -18,6 +18,7 @@ import { summariseCheckFailure } from "../src/checks/summary.ts";
18	18	import { SLOW_SECONDS, summaryLines } from "../scripts/gate-summary.ts";
19	19	import { tally } from "../scripts/shards.ts";
20	20	import {
	21	holdsLastSlot,
21	22	pickNext,
22	23	requeueAbandonedChecks,
23	24	startChecks,
		@@ -832,6 +833,15 @@ describe("checks", () => {
832	833	).toBe(true);
833	834	}, 60_000);
834	835
	836	test("a queued gate keeps the last free slot from pull checks", () => {
	837	// One slot free, a gate queued: the pull check passes, the gate starts.
	838	expect(holdsLastSlot(1, true)).toBe(true);
	839	// With no gate queued the pull check takes it, as every slot.
	840	expect(holdsLastSlot(1, false)).toBe(false);
	841	// More than one free: the pull check starts, one is still left.
	842	expect(holdsLastSlot(2, true)).toBe(false);
	843	});
	844
835	845	test("pickNext takes the newest queued check, unless one has starved", () => {
836	846	const a = createRun(app.db, { repoId: repo.id, kind: "check", env: "a" });
837	847	const b = createRun(app.db, { repoId: repo.id, kind: "check", env: "b" });
		@@ -0,0 +1,193 @@
	1	/**
	2	* The last free check slot belongs to a merge-queue gate (#6376), from the
	3	* time its candidate forms until its verdict: a pull's check waits for it,
	4	* a rebase's check and main's check do not, and with no gate every slot is
	5	* the checks'. The gates are real `enqueue` calls; the service schedules
	6	* once as it starts, and the tests wait on a run's state or a gate's own
	7	* promise, never on a fixed delay.
	8	*/
	9	import { afterAll, beforeAll, 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 { commitToBranch, createRepo } from "../src/actions/repos.ts";
	15	import { openPull } from "../src/actions/pulls.ts";
	16	import { type Candidate, enqueue, gateQueued, gatesInFlight, type Queued } from "../src/checks/queue.ts";
	17	import { startChecks } from "../src/checks/service.ts";
	18	import type { Pull } from "../src/core/pulls.ts";
	19	import type { Repo } from "../src/core/repos.ts";
	20	import { cancelRun, createRun, getRun, type Run } from "../src/core/runs.ts";
	21	import { createUser, type User } from "../src/core/users.ts";
	22	import { serve, type Servers } from "../src/http/server.ts";
	23
	24	let dataDir: string;
	25	let app: App;
	26	let servers: Servers;
	27	let alice: User;
	28	let repo: Repo;
	29	let pulls = 0;
	30
	31	beforeAll(async () => {
	32	dataDir = await mkdtemp(join(tmpdir(), "forge-gate-last-slot-"));
	33	app = createApp({ dataDir, port: 0 });
	34	servers = await serve(app);
	35	const a = await createUser(app.db, {
	36	origin: "seed",
	37	trusted: true,
	38	username: "alice",
	39	email: "[email protected]",
	40	password: "a-good-passphrase",
	41	adminIfFirst: true,
	42	});
	43	if (!a.ok) throw new Error(a.error);
	44	alice = a.value;
	45	const made = await createRepo(app, alice, { name: "lib", visibility: "public" });
	46	if (!made.ok) throw new Error(String(made.error));
	47	repo = made.value;
	48	const seeded = await commitToBranch(app, alice, repo, {
	49	message: "Start",
	50	files: [{ path: "README.md", content: "# lib\n" }],
	51	});
	52	if (!seeded.ok) throw new Error(JSON.stringify(seeded.error));
	53	});
	54
	55	afterAll(async () => {
	56	servers.stop();
	57	app.close();
	58	await rm(dataDir, { recursive: true, force: true });
	59	});
	60
	61	async function newPull(): Promise<Pull> {
	62	const branch = `b${++pulls}`;
	63	const branched = await commitToBranch(app, alice, repo, {
	64	branch,
	65	message: `Add ${branch}`,
	66	files: [{ path: `${branch}.md`, content: "ok\n" }],
	67	});
	68	if (!branched.ok) throw new Error(JSON.stringify(branched.error));
	69	const opened = await openPull(app, alice, repo, { title: branch, baseBranch: "main", headBranch: branch });
	70	if (!opened.ok) throw new Error(String(opened.error));
	71	return opened.value;
	72	}
	73
	74	const pullCheck = (pull: Pull): Run =>
	75	createRun(app.db, { repoId: repo.id, kind: "check", env: `pull-${pull.number}`, issueId: pull.id, commitOid: pull.headOid });
	76
	77	/** A gate whose candidate cannot be built: it is forming now, and over in a moment. */
	78	function formingGate(pull: Pull): Promise<boolean[]> {
	79	const entry: Queued = {
	80	pull,
	81	dir: join(dataDir, "nowhere"),
	82	land: async () => undefined,
	83	release: () => {},
	84	handBack: () => {},
	85	};
	86	const answer = enqueue(`gate-${pull.id}`, [entry], async () => ({ lands: false, interrupted: true }));
	87	expect(gateQueued()).toBe(true);
	88	return answer;
	89	}
	90
	91	/** Poll until `ok`, giving up at a generous deadline rather than asserting a speed. */
	92	async function until(ok: () => boolean): Promise<boolean> {
	93	for (let i = 0; i < 2000 && !ok(); i++) await Bun.sleep(5);
	94	return ok();
	95	}
		sentinelcrew 19d ago outdated This is a wall-clock wait. The "not queued" cases pass only if the service starts the run within 300 ms, which can flake on a loaded check machine. Wait on the state change instead. Also add the case that matters: a gate that is still running keeps the slot.
	96
	97	const state = (run: Run): Run["state"] => getRun(app.db, run.id)!.state;
	98
	99	/** Whether `run` leaves the queue once the service, with one slot, has had its turn. */
	100	async function starts(run: Run, gate: Promise<boolean[]> \| undefined): Promise<boolean> {
	101	const checks = startChecks(app, { forgeUrl: servers.url, concurrency: 1 });
	102	try {
	103	await gate;
	104	return await until(() => state(run) !== "queued");
	105	} finally {
	106	await checks.stop({ drain: false });
	107	cancelRun(app.db, run.id);
	108	}
	109	}
	110
	111	test("a pull's check does not take the last slot while a gate is forming", async () => {
	112	const run = pullCheck(await newPull());
	113	const gate = formingGate(await newPull());
	114	const checks = startChecks(app, { forgeUrl: servers.url, concurrency: 1 });
	115	try {
	116	// The service scheduled as it started, with the gate's candidate forming.
	117	expect(state(run)).toBe("queued");
	118	await gate;
	119	expect(await until(() => state(run) !== "queued")).toBe(true);
	120	} finally {
	121	await checks.stop({ drain: false });
	122	cancelRun(app.db, run.id);
	123	}
	124	expect(gateQueued()).toBe(false);
	125	});
	126
	127	test("a gate that is running keeps the slot: the pull's check starts when it settles", async () => {
	128	const dir = await mkdtemp(join(tmpdir(), "forge-gate-last-slot-git-"));
	129	const git = (...args: string[]) =>
	130	Bun.spawnSync(["git", "-C", dir, "-c", "user.name=t", "-c", "user.email=t@t", ...args]).stdout.toString().trim();
	131	Bun.spawnSync(["git", "init", "-q", "-b", "main", dir]);
	132	git("commit", "-q", "--allow-empty", "-m", "start");
	133	git("checkout", "-q", "-b", "feature");
	134	await Bun.write(join(dir, "f.txt"), "f\n");
	135	git("add", "f.txt");
	136	git("commit", "-q", "-m", "f");
	137	const headOid = git("rev-parse", "HEAD");
	138	git("checkout", "-q", "--detach", "main");
	139
	140	const gated = Promise.withResolvers<void>();
	141	let running = false;
	142	const entry: Queued = {
	143	pull: { id: "gate-running", number: 9999, baseBranch: "main", headOid } as Pull,
	144	dir,
	145	land: async () => undefined,
	146	release: () => {},
	147	handBack: () => {},
	148	};
	149	const answer = enqueue("gate-running-line", [entry], async (_candidate: Candidate) => {
	150	running = true;
	151	await gated.promise;
	152	return { lands: false, interrupted: true };
	153	});
	154	const run = pullCheck(await newPull());
	155	let checks: ReturnType<typeof startChecks> \| undefined;
	156	try {
	157	expect(await until(() => running)).toBe(true);
	158	// The candidate is built and the gate runs: nothing is forming, a gate is in flight.
	159	expect(gateQueued()).toBe(false);
	160	expect(gatesInFlight()).toBe(1);
	161	checks = startChecks(app, { forgeUrl: servers.url, concurrency: 1 });
	162	// The service scheduled as it started, and the one slot is the gate's.
	163	expect(state(run)).toBe("queued");
	164	await Bun.sleep(0);
	165	expect(state(run)).toBe("queued");
	166	gated.resolve();
	167	await answer;
	168	expect(await until(() => state(run) !== "queued")).toBe(true);
	169	} finally {
	170	gated.resolve();
	171	await answer;
	172	await checks?.stop({ drain: false });
	173	cancelRun(app.db, run.id);
	174	await rm(dir, { recursive: true, force: true });
	175	}
	176	expect(gatesInFlight()).toBe(0);
	177	});
	178
	179	test("with no gate forming, the pull's check takes the slot", async () => {
	180	const run = pullCheck(await newPull());
	181	expect(gateQueued()).toBe(false);
	182	expect(await starts(run, undefined)).toBe(true);
	183	});
	184
	185	test("a rebase's check and main's check still start while a gate is forming", async () => {
	186	const pull = await newPull();
	187	createRun(app.db, { repoId: repo.id, kind: "agent", issueId: pull.id, reason: "conflict" });
	188	const rebase = pullCheck(pull);
	189	expect(await starts(rebase, formingGate(await newPull()))).toBe(true);
	190
	191	const main = createRun(app.db, { repoId: repo.id, kind: "check", env: "main", commitOid: pull.headOid });
	192	expect(await starts(main, formingGate(await newPull()))).toBe(true);
	193	});