3f0ad2a · 44b2ac7
@@ -167,8 +167,14 @@
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.
A candidate carries at most `queue.batch` pulls (Settings → Capacity,
`FORGE_QUEUE_BATCH`, 1–32, default 8); the rest wait their turn. Each line
adapts below it (#6674): a red batch halves its size (`[queue] batch halves
to N`), two green batches in a row double it back (`[queue] batch grows to
N`), never past the setting, so a clean line drains in big batches and a
flaky one stays small. Pulls that come back to the line from a batch that
gated go again together, whatever the size. The tick line's `batch=` says
the size: `8`, or `4 of 16` after a red.
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,
@@ -178,6 +184,11 @@
gated tree. Red is searched: prefixes of the candidate are gated on main
— the front half, plus one more cut per free check slot, all at once — the
longest green prefix lands, the search goes on in the steps up to the first
red prefix, and the red pull is handed back with the failing tests named.
However few slots are free, the search cuts at least √n − 1 prefixes, and
a red run of three or more is then searched by leaving one out: one gate
per pull of the run, each carrying everything not yet landed but that pull.
The green one lands, and the pull it left out is gated alone. One bad pull
in 16 costs two rounds: 3 prefixes, then 4 gates.
A red batch is never formed again around the same head: with a check slot
free, the head of a red batch of three or more is gated alone beside the
line (`[queue] red batch: its head is gated alone`) while the rest gates
@@ -134,12 +134,18 @@
type Schedule = (fire: () => void, ms: number) => ReturnType<typeof setTimeout>;
let schedule: Schedule = setTimeout;
/**
 * 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.
 * The most pulls one candidate carries when the line is not told (#3430):
 * eight keep a gate near one full tier. The rest of the line waits for the
 * next candidate, in order. A forge sets it as `queue.batch` (#6674).
 */
export const MAX_CANDIDATE = 8;
/** The largest batch a line takes, whatever it is told (`queue.batch`'s range). */
export const MAX_BATCH = 32;
/** Green gates in a row that double a shrunk batch (#6674). */
const GREENS_TO_GROW = 2;
/** Candidates gating at once by default: one on main, the next on top of it. */
export const DEFAULT_DEPTH = 2;
@@ -147,6 +153,13 @@
export const DEFAULT_DEPTH = 2;
export interface QueueOptions {
  /** How many candidates may gate at once, each on the one before. */
  readonly depth?: number;
  /**
   * The most pulls a candidate carries (`queue.batch`, #6674): the line
   * starts here, halves after a red batch and doubles back after two green
   * ones in a row, never past it. Read on every enqueue, so a sweep brings a
   * change to a line already draining.
   */
  readonly batch?: number;
  /**
   * Check slots free now for a gate beyond a line's first, the pull
   * checks' reserve already kept back; none when not said.
@@ -301,6 +314,17 @@
interface Line {
  readonly suspects: Map<string, Suspect>;
  /** The last hour's candidate verdicts, search prefixes and re-checks not counted. */
  batches: BatchVerdict[];
  /** The most pulls a batch may carry, as the last enqueue was told (#6674). */
  ceiling: number;
  /** The batch size after a red, below the ceiling; undefined when at it (#6674). */
  size: number | undefined;
  /** Green batch verdicts in a row since the last red or the last doubling. */
  greens: number;
  /**
   * Pulls back in the line from a batch that gated: they gate together
   * again up to the ceiling, past a size a red halved (#6674).
   */
  readonly back: Set<Waiting>;
  draining: boolean;
  /** Wakes a drain waiting on a gate when pulls join the line. */
  wake: (() => void) | undefined;
@@ -405,6 +429,10 @@
function lineOf(repoId: string, store: QueueStore | undefined): Line {
      reds,
      suspects: new Map(),
      batches: [],
      ceiling: MAX_CANDIDATE,
      size: undefined,
      greens: 0,
      back: new Set(),
      draining: false,
      wake: undefined,
      store,
@@ -469,6 +497,7 @@
export async function enqueue(
  options: QueueOptions = {},
): Promise<boolean[]> {
  const line = lineOf(repoId, options.store);
  line.ceiling = batchCeiling(options.batch);
  // A red mark names the base it went red on; the tip is read only when a
  // mark is there to compare it with.
  const tips = new Map<string, string | undefined>();
@@ -485,20 +514,22 @@
export async function enqueue(
    if (red !== undefined && !held) clearRed(line, entry.pull.id);
    const recheck = held;
    if (line.suspects.get(entry.pull.id)?.head !== entry.pull.headOid) line.suspects.delete(entry.pull.id);
    return new Promise<boolean>((resolve) =>
      line.waiting.push({
    return new Promise<boolean>((resolve) => {
      const waiting: Waiting = {
        entry,
        recheck,
        settle: (landed) => {
          line.holds.delete(entry.pull.id);
          line.back.delete(waiting);
          if (landed) {
            clearRed(line, entry.pull.id);
            line.suspects.delete(entry.pull.id);
          }
          resolve(landed);
        },
      }),
    );
      };
      line.waiting.push(waiting);
    });
  });
  if (!line.draining) void drain(line, gate, options);
  else {
@@ -508,6 +539,49 @@
export async function enqueue(
  return Promise.all(answers);
}
/** `queue.batch` as a line takes it: a whole number from 1 to `MAX_BATCH`, else the default. */
function batchCeiling(batch: number | undefined): number {
  if (batch === undefined || !Number.isFinite(batch)) return MAX_CANDIDATE;
  return Math.max(1, Math.min(MAX_BATCH, Math.floor(batch)));
}
/** How many pulls the line's next batch carries: its adapted size, never past the ceiling. */
function batchSize(line: Line): number {
  return Math.min(line.ceiling, line.size ?? line.ceiling);
}
/**
 * A batch verdict moves the size (#6674): a red halves it, never below one;
 * two greens in a row double a shrunk one, back to the ceiling at most. A
 * clean line drains in big batches, a flaky one stays small.
 */
function adapt(line: Line, red: boolean): void {
  const was = batchSize(line);
  if (red) {
    line.greens = 0;
    line.size = Math.max(1, Math.floor(was / 2));
  } else {
    line.greens++;
    if (line.greens < GREENS_TO_GROW) return;
    line.greens = 0;
    if (line.size === undefined) return;
    line.size = line.size * 2 >= line.ceiling ? undefined : line.size * 2;
  }
  const now = batchSize(line);
  if (now !== was) log.info(`batch ${now > was ? "grows" : "halves"} to ${now}`, { repo: line.repoId, from: was });
}
/**
 * The size of a repository's next batch, for the tick line (#6674): `8`, or
 * `4 of 16` while a red keeps it below `ceiling`, the setting in force.
 */
export function batchWords(repoId: string, ceiling: number): string {
  const line = lines.get(repoId);
  const top = batchCeiling(ceiling);
  const size = Math.min(top, line?.size ?? top);
  return size < top ? `${size} of ${top}` : String(top);
}
/** The pulls waiting in a repository's line, not yet in a candidate: as they joined. */
export function waitingPulls(repoId: string): Pull[] {
  return (lines.get(repoId)?.waiting ?? []).map((w) => w.entry.pull);
@@ -789,7 +863,7 @@
const ARRIVED = Symbol("arrived");
 * first as its verdict comes. 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, and is not stacked on another. A batch
 * carries at most `MAX_CANDIDATE` pulls. A gate beyond the line's first
 * carries at most the line's batch size (`batchSize`). A gate beyond the line's first
 * starts only on a free check slot (`room`). A suspect is gated alone,
 * beside the line and first (#4999); the line's first gate waits for a
 * free slot while one is.
@@ -852,6 +926,7 @@
async function drain(line: Line, gate: GateCandidate, options: QueueOptions): Pr
        left = await discard(ahead);
      }
      // What waits for the next candidate goes back to the front.
      for (const w of left) line.back.add(w);
      line.waiting = [...left, ...line.waiting];
    }
  } finally {
@@ -896,6 +971,7 @@
async function solo(line: Line, gated: Gated, w: Waiting): Promise<{ readonly do
      if (line.stopped) return w.settle(false);
      if (verdict.lands) {
        line.suspects.set(pull.id, { head: pull.headOid, reds: 0, alone: false, cleared: true });
        line.back.add(w);
        line.waiting.unshift(w);
        return;
      }
@@ -975,9 +1051,12 @@
async function launch(line: Line, gated: Gated, on: Flight | undefined): Promise
  const branch = first.pull.baseBranch;
  if (on !== undefined && (on.recheck || first.dir !== on.candidate.dir || branch !== on.branch)) return "apart";
  const same = open.filter((w) => !w.recheck && w.entry.dir === first.dir && w.entry.pull.baseBranch === branch);
  // A halved size holds fresh pulls back; what came back from a batch gates together.
  const size = Math.min(line.ceiling, Math.max(batchSize(line), same.filter((w) => line.back.has(w)).length));
  const batch = front.recheck
    ? [front]
    : [...same.filter((w) => !cleared(w)), ...same.filter(cleared)].slice(0, MAX_CANDIDATE);
    : [...same.filter((w) => !cleared(w)), ...same.filter(cleared)].slice(0, size);
  for (const w of batch) line.back.delete(w);
  line.waiting = line.waiting.filter((w) => !batch.includes(w));
  let candidate: Candidate | undefined;
  line.forming++;
@@ -1092,6 +1171,7 @@
async function settleHead(search: Search, head: Flight, verdict: LandVerdict, ah
    }
  }
  if (verdict.lands) {
    if (!head.recheck) adapt(search.line, false);
    if (await land(candidate, head.steps)) return [];
    return discard(ahead);
  }
@@ -1109,6 +1189,9 @@
async function settleHead(search: Search, head: Flight, verdict: LandVerdict, ah
  const split =
    !head.recheck && head.steps.length > 2 && suspectOf(line, head.steps[0]!) === undefined && search.room() > 0;
  suspected(line, head.steps, split);
  // A red halves the batch (#6674) — but not a split: its rest gates again
  // as the one batch it was, and halves the size if it is red too.
  if (!head.recheck && !split) adapt(line, true);
  if (split) {
    log.info("red batch: its head is gated alone", {
      repo: line.repoId,
@@ -1206,6 +1289,14 @@
async function round(
 * red one wait, in line order, for the next candidate. No green verdict is
 * thrown away: a disjoint segment gated on the base alone was void once
 * another landed (#6353).
 *
 * However few slots are free, a search gates at least √n − 1 prefixes, so
 * the red run holds at most √n steps (#6674): a red batch blocks its line
 * and the flights stacked on it were stopped, so their slots are the
 * search's. That run is then searched by leaving one out (`leftOut`): one
 * gate per step of it, each carrying everything the search has not landed
 * but that step. With one bad pull, 16 red ones land their 15 good in two
 * rounds: 3 prefixes, then 4 gates.
 */
async function judged(
  search: Search,
@@ -1233,10 +1324,11 @@
async function judged(
    return [];
  }
  const n = steps.length;
  // One gate per free slot and one more, at least the front half, never the
  // whole: it is known red.
  // One gate per free slot and one more, at least the front half and √n − 1
  // prefixes, never the whole: it is known red.
      
        junipercrew
        18d ago
        
      
      Width can exceed free slots by up to ~5 at n=32. Cap it by the reserve kept for pull checks, or prove in a test that the reserve survives.
    
  const free = Math.max(0, room());
  const runs = split(steps, Math.min(n - 1, 1 + free));
  const width = Math.min(n - 1, Math.max(1 + free, Math.ceil(Math.sqrt(n)) - 1));
  const runs = split(steps, width);
  const ends: number[] = [];
  for (const run of runs) ends.push((ends.at(-1) ?? 0) + run.length);
  const prefixes = ends.slice(0, -1).map((end) => prefixOf(candidate, end));
@@ -1253,10 +1345,60 @@
async function judged(
  if (green > 0) await land(prefixes[bad - 1]!, steps.slice(0, green));
  const now = [...landed, ...steps.slice(0, green).map((w) => w.entry)];
  const after = runs.slice(bad + 1).flat();
  const left = await round(search, runs[bad]!, known, now);
  const suspects = runs[bad]!;
  // Leaving one out pays when the run is longer than two: of two, the front
  // one gated alone settles both, in line order.
  if (suspects.length > 2 && suspects.length <= width + 1) {
    const left = await leftOut(search, suspects, after);
    if (left !== undefined) return left;
  }
  const left = await round(search, suspects, known, now);
  return [...left, ...after];
}
/**
 * The red run `suspects` searched by leaving one out (#6674): for each of
 * its steps, everything left of the search but that step — the rest of the
 * run, then `after` — built on the base's tip, all gated at once. The first
 * green one lands, and the step it left out is a suspect gated alone beside
 * the line (#4999), handed back on its own red. Undefined when none is
 * green (more than one bad pull): the run is searched by prefixes as
 * before. Each answer is what waits for the next candidate.
 */
async function leftOut(search: Search, suspects: readonly Waiting[], after: readonly Waiting[]): Promise<Waiting[] | undefined> {
  const { line, gated } = search;
  const tries: { out: Waiting; batch: Waiting[]; candidate: Candidate }[] = [];
  for (const out of suspects) {
    const batch = [...suspects.filter((w) => w !== out), ...after];
    const candidate = await build(batch);
    if (candidate !== undefined && candidate.steps.length > 0) tries.push({ out, batch, candidate });
  }
  const verdicts = await Promise.all(tries.map((t) => gated(t.candidate)));
  if (line.stopped || verdicts.some((v) => !v.lands && "interrupted" in v)) {
    for (const w of [...suspects, ...after]) w.settle(false);
    return [];
  }
  const found = verdicts.findIndex((v) => v.lands);
  if (found < 0) return undefined;
  const { out, batch, candidate } = tries[found]!;
  log.info("red run: every pull but one lands, the one left out is gated alone", {
    repo: line.repoId,
    pull: out.entry.pull.number,
    landing: candidate.steps.length,
  });
  const was = suspectOf(line, out);
  line.suspects.set(out.entry.pull.id, {
    head: out.entry.pull.headOid,
    reds: was?.reds ?? 1,
    alone: true,
    cleared: false,
  });
  const steps = inCandidate(batch, candidate);
  // A step that did not land was settled by `land`; the one left out waits.
  await land(candidate, steps);
  return [out];
}
/** The candidate's chain cut after its first `end` steps: the tree a prefix gates. */
function prefixOf(candidate: Candidate, end: number): Candidate {
  const last = candidate.steps[end - 1]!;
@@ -20,6 +20,7 @@
export type CapacityKey =
  | "checks.concurrency"
  | "checks.shards"
  | "queue.depth"
  | "queue.batch"
  | "children.max"
  | "cells.egress"
  | "guard_new_accounts"
@@ -72,6 +73,13 @@
export const CAPACITY: Record<CapacityKey, Definition> = {
    fallback: "2",
    help: "How many merge-queue candidates of one repository gate at once, each on top of the one before: while one gates on main the next gates on it, and lands right after it if both are green. 1 gates one candidate at a time. A gate beyond the first takes a free check slot only, never one of the two kept for pull checks.",
  },
  "queue.batch": {
    label: "Merge queue batch size",
    env: "FORGE_QUEUE_BATCH",
    kind: { type: "count", min: 1, max: 32 },
    fallback: "8",
    help: "The most pulls one merge-queue candidate carries. A bigger batch lands more pulls per green gate; a red one costs more search gates to find its culprit (one bad pull in 16 is found in two rounds, at most √16 = 4 gates each). The queue starts each line here, halves the batch after a red gate and doubles it back after two green gates in a row, never past this. Read on every sweep.",
  },
  "children.max": {
    label: "Forges this one may send",
    env: "FORGE_CHILDREN_MAX",
@@ -318,6 +326,11 @@
export function queueDepth(db: Database): number {
  return Number(capacityValue(db, "queue.depth"));
}
/** The most pulls one merge-queue candidate carries (#6674); the line adapts below it. */
export function queueBatch(db: Database): number {
  return Number(capacityValue(db, "queue.batch"));
}
/** Whether the platform edge's identity is read at all; on every request. */
export function edgeSigninOn(db: Database): boolean {
  return capacityValue(db, "signin.edge") !== "off";
@@ -25,9 +25,9 @@
import type { AccessError } from "../actions/access.ts";
import { adoptGate, gateCandidate, releaseOnLand } from "../checks/landing.ts";
import { queueStore } from "../checks/queue_store.ts";
import { addLearned } from "../checks/learned.ts";
import { dropFromLine, enqueue, type GateCandidate, type Queued, type QueueOptions, queueClashes, queueWords, waitingPulls } from "../checks/queue.ts";
import { batchWords, dropFromLine, enqueue, type GateCandidate, type Queued, type QueueOptions, queueClashes, queueWords, waitingPulls } from "../checks/queue.ts";
import { checksFor } from "../checks/service.ts";
import { queueDepth } from "../core/capacity.ts";
import { queueBatch, queueDepth } from "../core/capacity.ts";
import { readConflictLane, readConflictRoom } from "./conflict_room.ts";
import { mergeable, resolveOid, treeOf } from "../git/merge.ts";
import { CLOSED_NOTE } from "../core/done_on_main.ts";
@@ -1321,6 +1321,8 @@
export async function autonomyTick(
        ...(approved === undefined ? {} : { approved }),
        // What the merge queue is doing now, not what the sweep admitted (#3430).
        queue: queueWords(current.id),
        // The size of the next batch: `8`, or `4 of 16` after a red (#6674).
        batch: batchWords(current.id, queueBatch(app.db)),
        // Pulls the queue found red on top of others of their candidate (#4235).
        clashes: queueClashes(current.id),
        spent: `${status.spentTodayUsd.toFixed(2)}/${status.ceilingUsd}`,
@@ -2053,14 +2055,15 @@
export function regateDue(red: Decision, now: number): boolean {
}
/**
 * How the merge queue may use the forge's check slots (#4053): its depth as
 * an admin set it, and the slots the checks service has spare. Each judged
 * How the merge queue may use the forge's check slots (#4053): its depth and
 * batch size as an admin set them, and the slots the checks service has spare. Each judged
 * batch is kept for the red-batch rate (#6427); its gates and red marks are
 * kept in the database, and a gate a roll left running adopted (#6611).
 */
function queueOptions(app: App, repo: Repo): QueueOptions {
  return {
    depth: queueDepth(app.db),
    batch: queueBatch(app.db),
    room: () => checksFor(app)?.spareSlots?.() ?? 0,
    onBatch: (repoId, red) => keepBatch(app.db, repoId, red),
    store: queueStore(app.db),
@@ -0,0 +1,148 @@
/**
 * The batch size (#6674): `queue.batch` caps how many pulls a candidate
 * carries; a red batch halves the line's size and two greens in a row
 * double it back; and a red batch of 16 with one bad pull lands its 15 good
 * in two rounds — √16 − 1 prefixes, then one gate per pull of the red run,
 * each leaving that pull out. The gate is held open and let go a wave at a
 * time, so "at once" is counted, never timed; it reads the candidate's tree:
 * red when it holds `bad.txt`.
 */
import { afterAll, beforeAll, describe, expect, test } from "bun:test";
import { mkdtemp, rm } from "node:fs/promises";
import { tmpdir } from "node:os";
import { join } from "node:path";
import { batchWords, type Candidate, enqueue, type Expect, type Queued } from "../src/checks/queue.ts";
import type { LandVerdict } from "../src/checks/tiers.ts";
import type { Pull } from "../src/core/pulls.ts";
describe("the merge queue's batch size", () => {
  let dir: string;
  const told: number[] = [];
  let landings = 0;
  const git = (...args: string[]) =>
    Bun.spawnSync(["git", "-C", dir, "-c", "user.name=t", "-c", "user.email=t@t", ...args]).stdout.toString().trim();
  beforeAll(async () => {
    dir = await mkdtemp(join(tmpdir(), "forge-merge-queue-batch-"));
    Bun.spawnSync(["git", "init", "-q", "-b", "main", dir]);
    git("commit", "-q", "--allow-empty", "-m", "start");
  });
  afterAll(async () => {
    await rm(dir, { recursive: true, force: true });
  });
  async function headAdding(branch: string, file: string): Promise<string> {
    git("checkout", "-q", "-B", branch, "main");
    git("reset", "-q", "--hard", "main");
    await Bun.write(join(dir, file), `${file}\n`);
    git("add", file);
    git("commit", "-q", "-m", file);
    git("checkout", "-q", "--detach");
    return git("rev-parse", branch);
  }
  const verdictOf = (candidate: Candidate): LandVerdict =>
    git("ls-tree", "--name-only", candidate.tipOid).split("\n").includes("bad.txt")
      ? { lands: false, reopen: true, comment: "red", learn: [] }
      : { lands: true };
  function entry(line: string, number: number, headOid: string): Queued {
    return {
      pull: { id: `${line}-${number}`, number, baseBranch: "main", headOid } as Pull,
      dir,
      land: async (expect: Expect) => {
        if (git("rev-parse", "refs/heads/main") !== expect.baseOid) return undefined;
        const tip = git("commit-tree", expect.treeOid, "-p", expect.baseOid, "-m", `#${number}`);
        git("update-ref", "refs/heads/main", tip);
        landings++;
        return tip;
      },
      release: () => {},
      handBack: () => told.push(number),
    };
  }
  async function pulls(line: string, from: number, count: number, bad?: number): Promise<Queued[]> {
    const out: Queued[] = [];
    for (let n = from; n < from + count; n++)
      out.push(entry(line, n, await headAdding(`${line}${n}`, n === bad ? "bad.txt" : `${line}${n}.txt`)));
    return out;
  }
  /** A gate that answers at once and keeps how many pulls each candidate carried. */
  function counting() {
    const sizes: number[] = [];
    const gate = async (candidate: Candidate) => {
      sizes.push(candidate.steps.length);
      return verdictOf(candidate);
    };
    return { gate, sizes };
  }
  test("the setting is honoured: 4 a candidate when told 4, 16 when told 16, never past 32", async () => {
    const four = counting();
    expect(await enqueue("told-four", await pulls("f", 1, 10), four.gate, { depth: 1, batch: 4 })).toEqual(
      Array(10).fill(true),
    );
    expect(four.sizes).toEqual([4, 4, 2]);
    const sixteen = counting();
    await enqueue("told-sixteen", await pulls("s", 1, 20), sixteen.gate, { depth: 1, batch: 16 });
    expect(sixteen.sizes).toEqual([16, 4]);
    expect(batchWords("told-sixteen", 16)).toBe("16");
    expect(batchWords("told-sixteen", 99)).toBe("32");
  });
  test("a red halves the batch, and two greens in a row double it back to the setting", async () => {
    told.length = 0;
    const { gate, sizes } = counting();
    // Eight, the first bad: red. The search finds it; the seven land.
    const first = await enqueue("adapts", await pulls("a", 1, 8, 1), gate, { depth: 1, batch: 8 });
    expect(first).toEqual([false, ...Array(7).fill(true)]);
    expect(told).toEqual([1]);
    expect(batchWords("adapts", 8)).toBe("4 of 8");
    // Sixteen good: two of four, both green, so the third is eight again.
    sizes.length = 0;
    expect(await enqueue("adapts", await pulls("a", 101, 16), gate, { depth: 1, batch: 8 })).toEqual(
      Array(16).fill(true),
    );
    expect(sizes).toEqual([4, 4, 8]);
    expect(batchWords("adapts", 8)).toBe("8");
  });
  test("a red 16-batch with one bad pull lands its 15 good within 2 rounds, then the bad one is handed back alone", async () => {
    told.length = 0;
    landings = 0;
    let pending: { candidate: Candidate; go: (v: LandVerdict) => void }[] = [];
    const seen: number[][][] = [];
    const gate = (candidate: Candidate) => new Promise<LandVerdict>((go) => pending.push({ candidate, go }));
    const letGo = async (n: number) => {
      for (let i = 0; i < 2000 && pending.length < n; i++) await Bun.sleep(5);
      expect(pending).toHaveLength(n);
      const going = pending;
      pending = [];
      seen.push(going.map((p) => p.candidate.steps.map((s) => s.entry.pull.number)));
      for (const p of going) p.go(verdictOf(p.candidate));
    };
    const line = await pulls("r", 1, 16, 6);
    // No free slot: the search still gates √16 − 1 = 3 prefixes.
    const landed = enqueue("red-sixteen", line, gate, { depth: 1, room: () => 0, batch: 16 });
    await letGo(1); // the sixteen: red
      
        junipercrew
        18d ago
        
      
      Polling loops with sleep are flake-prone. Await a gate-count promise resolved from the gate callback instead.
    
    await letGo(3); // round one: 4, 8 and 12 deep; the first four land, the red run is 5–8
    expect(seen[1]!.map((c) => c.length)).toEqual([4, 8, 12]);
    for (let i = 0; i < 2000 && pending.length < 4; i++) await Bun.sleep(5);
    expect(landings).toBe(4);
    await letGo(4); // round two: everything left but one of 5–8 each; the one without 6 lands
    expect(seen[2]!.map((c) => c.length)).toEqual([11, 11, 11, 11]);
    for (let i = 0; i < 2000 && landings < 15; i++) await Bun.sleep(5);
    expect(landings).toBe(15);
    expect(git("ls-tree", "--name-only", "refs/heads/main").split("\n").filter((f) => /^r\d+\.txt$/.test(f))).toHaveLength(15);
    await letGo(1); // the one left out, alone: red, handed back
    expect(seen[3]).toEqual([[6]]);
    expect(await landed).toEqual(line.map((q) => q.pull.number !== 6));
    expect(told).toEqual([6]);
    expect(git("ls-tree", "--name-only", "refs/heads/main").split("\n")).not.toContain("bad.txt");
    expect(batchWords("red-sixteen", 16)).toBe("8 of 16");
  });
});
@@ -11,7 +11,7 @@
import { afterAll, beforeAll, describe, expect, test } from "bun:test";
import { mkdtemp, rm } from "node:fs/promises";
import { tmpdir } from "node:os";
import { join } from "node:path";
import { type Candidate, enqueue, type Expect, queueClashes, type Queued, redBatchRate, redMark } from "../src/checks/queue.ts";
import { batchWords, type Candidate, enqueue, type Expect, queueClashes, type Queued, redBatchRate, redMark } from "../src/checks/queue.ts";
import type { Pull } from "../src/core/pulls.ts";
describe("the merge queue remembers clashes", () => {
@@ -177,12 +177,18 @@
describe("the merge queue remembers clashes", () => {
    expect(redMark("no-such-line", "clash-2", head)).toBeUndefined();
  });
  test("a push is a new head: B is forgotten, batched with the rest, and lands", async () => {
  test("a push is a new head: B is forgotten, gated as a fresh pull, and lands", async () => {
    const pushed = entry(2, await headAdding("b2", "b2.txt"));
    const g = entry(7, await headAdding("g", "g.txt"));
    const before = gates.length;
    expect(await enqueue(repoId, [pushed, g], gate)).toEqual([true, true]);
    expect(gates.slice(before)).toEqual([{ pulls: [2, 7], ok: true }]);
    // Not a re-check: B goes first, as fresh. Its red batches halved this
    // line's batch to one (#6674); two greens in a row double it.
    expect(gates.slice(before)).toEqual([
      { pulls: [2], ok: true },
      { pulls: [7], ok: true },
    ]);
    expect(batchWords(repoId, 8)).toBe("2 of 8");
    expect(queueClashes(repoId)).toBe(0);
  });
});
@@ -13,8 +13,9 @@
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, gateDeadlineForTest, MAX_CANDIDATE, gateRuns, queueStatus, queueWords, type Queued } from "../src/checks/queue.ts";
import { batchWords, enqueue, gateDeadlineForTest, MAX_CANDIDATE, gateRuns, queueStatus, queueWords, type Queued } from "../src/checks/queue.ts";
import { provideChecksForTest, type ChecksHandle } from "../src/checks/service.ts";
import { queueBatch, setCapacity, validCapacity } from "../src/core/capacity.ts";
import { latestDecision } from "../src/core/decisions.ts";
import { getPull, type Pull } from "../src/core/pulls.ts";
import { getRepoById, type Repo } from "../src/core/repos.ts";
@@ -225,6 +226,31 @@
describe("the merge queue", () => {
    expect(queue.some((l) => l.includes("[queue] gated 2 pulls: green"))).toBe(true);
  }, SLOW);
  test("queue.batch is read on each sweep: three per candidate when it says 3, and /settings/capacity explains the trade-off (#6674)", async () => {
    const repo = await autonomousRepo("batch-of-three");
    const base = git(repo, "rev-parse", "refs/heads/main");
    const pulls = [];
    for (let i = 0; i < 7; i++) pulls.push(await greenPull(repo, `b${i}.ts`));
    expect(validCapacity("queue.batch", "33")).toBe(false);
    expect(validCapacity("queue.batch", "0")).toBe(false);
    setCapacity(app.db, "queue.batch", "3", alice.id);
    try {
      expect(await sweepGreen(app, repo)).toBe(7);
    } finally {
      setCapacity(app.db, "queue.batch", null, alice.id);
    }
    const carried = gatesOf(repo)
      .map((g) => Number(git(repo, "rev-list", "--count", `${base}..${g.commit}`)))
      .sort((a, b) => a - b);
    expect(carried).toEqual([3, 6, 7]);
    for (const pull of pulls) expect(merged(repo, pull)).toBe(true);
    expect(batchWords(repo.id, queueBatch(app.db))).toBe("8");
    const page = await fetch(`${servers.url}/settings/capacity`, { headers: { cookie: await login() } });
    const html = await page.text();
    expect(html).toContain("Merge queue batch size");
    expect(html).toContain("a red one costs more search gates");
  }, SLOW);
  test("after a red batch, the good pulls go back into ONE candidate up to the cap, not one gate each (#4053)", async () => {
    const repo = await autonomousRepo("rebatch");
    const pulls = [];
		@@ -167,8 +167,14 @@ word) and `issue.parked = 1` carries it, so raw SQL never reads it as done.
167	167	its repository's queue (`src/checks/queue.ts`). The queue builds one
168	168	candidate: main, then each queued pull's squash on top of the one before, in
169	169	queue order, leaving out any pull that conflicts or has nothing left to land.
170		A candidate carries at most eight pulls (`MAX_CANDIDATE`), so a gate stays
171		near one full tier and a red bisects in three; the rest wait their turn.
	170	A candidate carries at most `queue.batch` pulls (Settings → Capacity,
	171	`FORGE_QUEUE_BATCH`, 1–32, default 8); the rest wait their turn. Each line
	172	adapts below it (#6674): a red batch halves its size (`[queue] batch halves
	173	to N`), two green batches in a row double it back (`[queue] batch grows to
	174	N`), never past the setting, so a clean line drains in big batches and a
	175	flaky one stays small. Pulls that come back to the line from a batch that
	176	gated go again together, whatever the size. The tick line's `batch=` says
	177	the size: `8`, or `4 of 16` after a red.
172	178	The log says `[queue] gating N pulls since HH:MM` as a gate starts and
173	179	`[queue] gated N pulls: green\|red\|cut off` as it ends.
174	180	The full gate runs once on that candidate. Green lands every pull in order,
		@@ -178,6 +184,11 @@ gated tree. Red is searched: prefixes of the candidate are gated on main
178	184	— the front half, plus one more cut per free check slot, all at once — the
179	185	longest green prefix lands, the search goes on in the steps up to the first
180	186	red prefix, and the red pull is handed back with the failing tests named.
	187	However few slots are free, the search cuts at least √n − 1 prefixes, and
	188	a red run of three or more is then searched by leaving one out: one gate
	189	per pull of the run, each carrying everything not yet landed but that pull.
	190	The green one lands, and the pull it left out is gated alone. One bad pull
	191	in 16 costs two rounds: 3 prefixes, then 4 gates.
181	192	A red batch is never formed again around the same head: with a check slot
182	193	free, the head of a red batch of three or more is gated alone beside the
183	194	line (`[queue] red batch: its head is gated alone`) while the rest gates
		@@ -134,12 +134,18 @@ type Schedule = (fire: () => void, ms: number) => ReturnType<typeof setTimeout>;
134	134	let schedule: Schedule = setTimeout;
135	135
136	136	/**
137		* The most pulls one candidate carries (#3430): eight keep a gate near one
138		* full tier and a red's bisection three gates deep. The rest of the line
139		* waits for the next candidate, in order.
	137	* The most pulls one candidate carries when the line is not told (#3430):
	138	* eight keep a gate near one full tier. The rest of the line waits for the
	139	* next candidate, in order. A forge sets it as `queue.batch` (#6674).
140	140	*/
141	141	export const MAX_CANDIDATE = 8;
142	142
	143	/** The largest batch a line takes, whatever it is told (`queue.batch`'s range). */
	144	export const MAX_BATCH = 32;
	145
	146	/** Green gates in a row that double a shrunk batch (#6674). */
	147	const GREENS_TO_GROW = 2;
	148
143	149	/** Candidates gating at once by default: one on main, the next on top of it. */
144	150	export const DEFAULT_DEPTH = 2;
145	151
		@@ -147,6 +153,13 @@ export const DEFAULT_DEPTH = 2;
147	153	export interface QueueOptions {
148	154	/** How many candidates may gate at once, each on the one before. */
149	155	readonly depth?: number;
	156	/**
	157	* The most pulls a candidate carries (`queue.batch`, #6674): the line
	158	* starts here, halves after a red batch and doubles back after two green
	159	* ones in a row, never past it. Read on every enqueue, so a sweep brings a
	160	* change to a line already draining.
	161	*/
	162	readonly batch?: number;
150	163	/**
151	164	* Check slots free now for a gate beyond a line's first, the pull
152	165	* checks' reserve already kept back; none when not said.
		@@ -301,6 +314,17 @@ interface Line {
301	314	readonly suspects: Map<string, Suspect>;
302	315	/** The last hour's candidate verdicts, search prefixes and re-checks not counted. */
303	316	batches: BatchVerdict[];
	317	/** The most pulls a batch may carry, as the last enqueue was told (#6674). */
	318	ceiling: number;
	319	/** The batch size after a red, below the ceiling; undefined when at it (#6674). */
	320	size: number \| undefined;
	321	/** Green batch verdicts in a row since the last red or the last doubling. */
	322	greens: number;
	323	/**
	324	* Pulls back in the line from a batch that gated: they gate together
	325	* again up to the ceiling, past a size a red halved (#6674).
	326	*/
	327	readonly back: Set<Waiting>;
304	328	draining: boolean;
305	329	/** Wakes a drain waiting on a gate when pulls join the line. */
306	330	wake: (() => void) \| undefined;
		@@ -405,6 +429,10 @@ function lineOf(repoId: string, store: QueueStore \| undefined): Line {
405	429	reds,
406	430	suspects: new Map(),
407	431	batches: [],
	432	ceiling: MAX_CANDIDATE,
	433	size: undefined,
	434	greens: 0,
	435	back: new Set(),
408	436	draining: false,
409	437	wake: undefined,
410	438	store,
		@@ -469,6 +497,7 @@ export async function enqueue(
469	497	options: QueueOptions = {},
470	498	): Promise<boolean[]> {
471	499	const line = lineOf(repoId, options.store);
	500	line.ceiling = batchCeiling(options.batch);
472	501	// A red mark names the base it went red on; the tip is read only when a
473	502	// mark is there to compare it with.
474	503	const tips = new Map<string, string \| undefined>();
		@@ -485,20 +514,22 @@ export async function enqueue(
485	514	if (red !== undefined && !held) clearRed(line, entry.pull.id);
486	515	const recheck = held;
487	516	if (line.suspects.get(entry.pull.id)?.head !== entry.pull.headOid) line.suspects.delete(entry.pull.id);
488		return new Promise<boolean>((resolve) =>
489		line.waiting.push({
	517	return new Promise<boolean>((resolve) => {
	518	const waiting: Waiting = {
490	519	entry,
491	520	recheck,
492	521	settle: (landed) => {
493	522	line.holds.delete(entry.pull.id);
	523	line.back.delete(waiting);
494	524	if (landed) {
495	525	clearRed(line, entry.pull.id);
496	526	line.suspects.delete(entry.pull.id);
497	527	}
498	528	resolve(landed);
499	529	},
500		}),
501		);
	530	};
	531	line.waiting.push(waiting);
	532	});
502	533	});
503	534	if (!line.draining) void drain(line, gate, options);
504	535	else {
		@@ -508,6 +539,49 @@ export async function enqueue(
508	539	return Promise.all(answers);
509	540	}
510	541
	542	/** `queue.batch` as a line takes it: a whole number from 1 to `MAX_BATCH`, else the default. */
	543	function batchCeiling(batch: number \| undefined): number {
	544	if (batch === undefined \|\| !Number.isFinite(batch)) return MAX_CANDIDATE;
	545	return Math.max(1, Math.min(MAX_BATCH, Math.floor(batch)));
	546	}
	547
	548	/** How many pulls the line's next batch carries: its adapted size, never past the ceiling. */
	549	function batchSize(line: Line): number {
	550	return Math.min(line.ceiling, line.size ?? line.ceiling);
	551	}
	552
	553	/**
	554	* A batch verdict moves the size (#6674): a red halves it, never below one;
	555	* two greens in a row double a shrunk one, back to the ceiling at most. A
	556	* clean line drains in big batches, a flaky one stays small.
	557	*/
	558	function adapt(line: Line, red: boolean): void {
	559	const was = batchSize(line);
	560	if (red) {
	561	line.greens = 0;
	562	line.size = Math.max(1, Math.floor(was / 2));
	563	} else {
	564	line.greens++;
	565	if (line.greens < GREENS_TO_GROW) return;
	566	line.greens = 0;
	567	if (line.size === undefined) return;
	568	line.size = line.size * 2 >= line.ceiling ? undefined : line.size * 2;
	569	}
	570	const now = batchSize(line);
	571	if (now !== was) log.info(`batch ${now > was ? "grows" : "halves"} to ${now}`, { repo: line.repoId, from: was });
	572	}
	573
	574	/**
	575	* The size of a repository's next batch, for the tick line (#6674): `8`, or
	576	* `4 of 16` while a red keeps it below `ceiling`, the setting in force.
	577	*/
	578	export function batchWords(repoId: string, ceiling: number): string {
	579	const line = lines.get(repoId);
	580	const top = batchCeiling(ceiling);
	581	const size = Math.min(top, line?.size ?? top);
	582	return size < top ? `${size} of ${top}` : String(top);
	583	}
	584
511	585	/** The pulls waiting in a repository's line, not yet in a candidate: as they joined. */
512	586	export function waitingPulls(repoId: string): Pull[] {
513	587	return (lines.get(repoId)?.waiting ?? []).map((w) => w.entry.pull);
		@@ -789,7 +863,7 @@ const ARRIVED = Symbol("arrived");
789	863	* first as its verdict comes. A batch is the front of the line that shares
790	864	* one place and one base branch; a sealed pull whose objects live
791	865	* elsewhere is a batch of its own, and is not stacked on another. A batch
792		* carries at most `MAX_CANDIDATE` pulls. A gate beyond the line's first
	866	* carries at most the line's batch size (`batchSize`). A gate beyond the line's first
793	867	* starts only on a free check slot (`room`). A suspect is gated alone,
794	868	* beside the line and first (#4999); the line's first gate waits for a
795	869	* free slot while one is.
		@@ -852,6 +926,7 @@ async function drain(line: Line, gate: GateCandidate, options: QueueOptions): Pr
852	926	left = await discard(ahead);
853	927	}
854	928	// What waits for the next candidate goes back to the front.
	929	for (const w of left) line.back.add(w);
855	930	line.waiting = [...left, ...line.waiting];
856	931	}
857	932	} finally {
		@@ -896,6 +971,7 @@ async function solo(line: Line, gated: Gated, w: Waiting): Promise<{ readonly do
896	971	if (line.stopped) return w.settle(false);
897	972	if (verdict.lands) {
898	973	line.suspects.set(pull.id, { head: pull.headOid, reds: 0, alone: false, cleared: true });
	974	line.back.add(w);
899	975	line.waiting.unshift(w);
900	976	return;
901	977	}
		@@ -975,9 +1051,12 @@ async function launch(line: Line, gated: Gated, on: Flight \| undefined): Promise
975	1051	const branch = first.pull.baseBranch;
976	1052	if (on !== undefined && (on.recheck \|\| first.dir !== on.candidate.dir \|\| branch !== on.branch)) return "apart";
977	1053	const same = open.filter((w) => !w.recheck && w.entry.dir === first.dir && w.entry.pull.baseBranch === branch);
	1054	// A halved size holds fresh pulls back; what came back from a batch gates together.
	1055	const size = Math.min(line.ceiling, Math.max(batchSize(line), same.filter((w) => line.back.has(w)).length));
978	1056	const batch = front.recheck
979	1057	? [front]
980		: [...same.filter((w) => !cleared(w)), ...same.filter(cleared)].slice(0, MAX_CANDIDATE);
	1058	: [...same.filter((w) => !cleared(w)), ...same.filter(cleared)].slice(0, size);
	1059	for (const w of batch) line.back.delete(w);
981	1060	line.waiting = line.waiting.filter((w) => !batch.includes(w));
982	1061	let candidate: Candidate \| undefined;
983	1062	line.forming++;
		@@ -1092,6 +1171,7 @@ async function settleHead(search: Search, head: Flight, verdict: LandVerdict, ah
1092	1171	}
1093	1172	}
1094	1173	if (verdict.lands) {
	1174	if (!head.recheck) adapt(search.line, false);
1095	1175	if (await land(candidate, head.steps)) return [];
1096	1176	return discard(ahead);
1097	1177	}
		@@ -1109,6 +1189,9 @@ async function settleHead(search: Search, head: Flight, verdict: LandVerdict, ah
1109	1189	const split =
1110	1190	!head.recheck && head.steps.length > 2 && suspectOf(line, head.steps[0]!) === undefined && search.room() > 0;
1111	1191	suspected(line, head.steps, split);
	1192	// A red halves the batch (#6674) — but not a split: its rest gates again
	1193	// as the one batch it was, and halves the size if it is red too.
	1194	if (!head.recheck && !split) adapt(line, true);
1112	1195	if (split) {
1113	1196	log.info("red batch: its head is gated alone", {
1114	1197	repo: line.repoId,
		@@ -1206,6 +1289,14 @@ async function round(
1206	1289	* red one wait, in line order, for the next candidate. No green verdict is
1207	1290	* thrown away: a disjoint segment gated on the base alone was void once
1208	1291	* another landed (#6353).
	1292	*
	1293	* However few slots are free, a search gates at least √n − 1 prefixes, so
	1294	* the red run holds at most √n steps (#6674): a red batch blocks its line
	1295	* and the flights stacked on it were stopped, so their slots are the
	1296	* search's. That run is then searched by leaving one out (`leftOut`): one
	1297	* gate per step of it, each carrying everything the search has not landed
	1298	* but that step. With one bad pull, 16 red ones land their 15 good in two
	1299	* rounds: 3 prefixes, then 4 gates.
1209	1300	*/
1210	1301	async function judged(
1211	1302	search: Search,
		@@ -1233,10 +1324,11 @@ async function judged(
1233	1324	return [];
1234	1325	}
1235	1326	const n = steps.length;
1236		// One gate per free slot and one more, at least the front half, never the
1237		// whole: it is known red.
	1327	// One gate per free slot and one more, at least the front half and √n − 1
	1328	// prefixes, never the whole: it is known red.
		junipercrew 18d ago Width can exceed free slots by up to ~5 at n=32. Cap it by the reserve kept for pull checks, or prove in a test that the reserve survives.
1238	1329	const free = Math.max(0, room());
1239		const runs = split(steps, Math.min(n - 1, 1 + free));
	1330	const width = Math.min(n - 1, Math.max(1 + free, Math.ceil(Math.sqrt(n)) - 1));
	1331	const runs = split(steps, width);
1240	1332	const ends: number[] = [];
1241	1333	for (const run of runs) ends.push((ends.at(-1) ?? 0) + run.length);
1242	1334	const prefixes = ends.slice(0, -1).map((end) => prefixOf(candidate, end));
		@@ -1253,10 +1345,60 @@ async function judged(
1253	1345	if (green > 0) await land(prefixes[bad - 1]!, steps.slice(0, green));
1254	1346	const now = [...landed, ...steps.slice(0, green).map((w) => w.entry)];
1255	1347	const after = runs.slice(bad + 1).flat();
1256		const left = await round(search, runs[bad]!, known, now);
	1348	const suspects = runs[bad]!;
	1349	// Leaving one out pays when the run is longer than two: of two, the front
	1350	// one gated alone settles both, in line order.
	1351	if (suspects.length > 2 && suspects.length <= width + 1) {
	1352	const left = await leftOut(search, suspects, after);
	1353	if (left !== undefined) return left;
	1354	}
	1355	const left = await round(search, suspects, known, now);
1257	1356	return [...left, ...after];
1258	1357	}
1259	1358
	1359	/**
	1360	* The red run `suspects` searched by leaving one out (#6674): for each of
	1361	* its steps, everything left of the search but that step — the rest of the
	1362	* run, then `after` — built on the base's tip, all gated at once. The first
	1363	* green one lands, and the step it left out is a suspect gated alone beside
	1364	* the line (#4999), handed back on its own red. Undefined when none is
	1365	* green (more than one bad pull): the run is searched by prefixes as
	1366	* before. Each answer is what waits for the next candidate.
	1367	*/
	1368	async function leftOut(search: Search, suspects: readonly Waiting[], after: readonly Waiting[]): Promise<Waiting[] \| undefined> {
	1369	const { line, gated } = search;
	1370	const tries: { out: Waiting; batch: Waiting[]; candidate: Candidate }[] = [];
	1371	for (const out of suspects) {
	1372	const batch = [...suspects.filter((w) => w !== out), ...after];
	1373	const candidate = await build(batch);
	1374	if (candidate !== undefined && candidate.steps.length > 0) tries.push({ out, batch, candidate });
	1375	}
	1376	const verdicts = await Promise.all(tries.map((t) => gated(t.candidate)));
	1377	if (line.stopped \|\| verdicts.some((v) => !v.lands && "interrupted" in v)) {
	1378	for (const w of [...suspects, ...after]) w.settle(false);
	1379	return [];
	1380	}
	1381	const found = verdicts.findIndex((v) => v.lands);
	1382	if (found < 0) return undefined;
	1383	const { out, batch, candidate } = tries[found]!;
	1384	log.info("red run: every pull but one lands, the one left out is gated alone", {
	1385	repo: line.repoId,
	1386	pull: out.entry.pull.number,
	1387	landing: candidate.steps.length,
	1388	});
	1389	const was = suspectOf(line, out);
	1390	line.suspects.set(out.entry.pull.id, {
	1391	head: out.entry.pull.headOid,
	1392	reds: was?.reds ?? 1,
	1393	alone: true,
	1394	cleared: false,
	1395	});
	1396	const steps = inCandidate(batch, candidate);
	1397	// A step that did not land was settled by `land`; the one left out waits.
	1398	await land(candidate, steps);
	1399	return [out];
	1400	}
	1401
1260	1402	/** The candidate's chain cut after its first `end` steps: the tree a prefix gates. */
1261	1403	function prefixOf(candidate: Candidate, end: number): Candidate {
1262	1404	const last = candidate.steps[end - 1]!;
		@@ -20,6 +20,7 @@ export type CapacityKey =
20	20	\| "checks.concurrency"
21	21	\| "checks.shards"
22	22	\| "queue.depth"
	23	\| "queue.batch"
23	24	\| "children.max"
24	25	\| "cells.egress"
25	26	\| "guard_new_accounts"
		@@ -72,6 +73,13 @@ export const CAPACITY: Record<CapacityKey, Definition> = {
72	73	fallback: "2",
73	74	help: "How many merge-queue candidates of one repository gate at once, each on top of the one before: while one gates on main the next gates on it, and lands right after it if both are green. 1 gates one candidate at a time. A gate beyond the first takes a free check slot only, never one of the two kept for pull checks.",
74	75	},
	76	"queue.batch": {
	77	label: "Merge queue batch size",
	78	env: "FORGE_QUEUE_BATCH",
	79	kind: { type: "count", min: 1, max: 32 },
	80	fallback: "8",
	81	help: "The most pulls one merge-queue candidate carries. A bigger batch lands more pulls per green gate; a red one costs more search gates to find its culprit (one bad pull in 16 is found in two rounds, at most √16 = 4 gates each). The queue starts each line here, halves the batch after a red gate and doubles it back after two green gates in a row, never past this. Read on every sweep.",
	82	},
75	83	"children.max": {
76	84	label: "Forges this one may send",
77	85	env: "FORGE_CHILDREN_MAX",
		@@ -318,6 +326,11 @@ export function queueDepth(db: Database): number {
318	326	return Number(capacityValue(db, "queue.depth"));
319	327	}
320	328
	329	/** The most pulls one merge-queue candidate carries (#6674); the line adapts below it. */
	330	export function queueBatch(db: Database): number {
	331	return Number(capacityValue(db, "queue.batch"));
	332	}
	333
321	334	/** Whether the platform edge's identity is read at all; on every request. */
322	335	export function edgeSigninOn(db: Database): boolean {
323	336	return capacityValue(db, "signin.edge") !== "off";
		@@ -25,9 +25,9 @@ import type { AccessError } from "../actions/access.ts";
25	25	import { adoptGate, gateCandidate, releaseOnLand } from "../checks/landing.ts";
26	26	import { queueStore } from "../checks/queue_store.ts";
27	27	import { addLearned } from "../checks/learned.ts";
28		import { dropFromLine, enqueue, type GateCandidate, type Queued, type QueueOptions, queueClashes, queueWords, waitingPulls } from "../checks/queue.ts";
	28	import { batchWords, dropFromLine, enqueue, type GateCandidate, type Queued, type QueueOptions, queueClashes, queueWords, waitingPulls } from "../checks/queue.ts";
29	29	import { checksFor } from "../checks/service.ts";
30		import { queueDepth } from "../core/capacity.ts";
	30	import { queueBatch, queueDepth } from "../core/capacity.ts";
31	31	import { readConflictLane, readConflictRoom } from "./conflict_room.ts";
32	32	import { mergeable, resolveOid, treeOf } from "../git/merge.ts";
33	33	import { CLOSED_NOTE } from "../core/done_on_main.ts";
		@@ -1321,6 +1321,8 @@ export async function autonomyTick(
1321	1321	...(approved === undefined ? {} : { approved }),
1322	1322	// What the merge queue is doing now, not what the sweep admitted (#3430).
1323	1323	queue: queueWords(current.id),
	1324	// The size of the next batch: `8`, or `4 of 16` after a red (#6674).
	1325	batch: batchWords(current.id, queueBatch(app.db)),
1324	1326	// Pulls the queue found red on top of others of their candidate (#4235).
1325	1327	clashes: queueClashes(current.id),
1326	1328	spent: `${status.spentTodayUsd.toFixed(2)}/${status.ceilingUsd}`,
		@@ -2053,14 +2055,15 @@ export function regateDue(red: Decision, now: number): boolean {
2053	2055	}
2054	2056
2055	2057	/**
2056		* How the merge queue may use the forge's check slots (#4053): its depth as
2057		* an admin set it, and the slots the checks service has spare. Each judged
	2058	* How the merge queue may use the forge's check slots (#4053): its depth and
	2059	* batch size as an admin set them, and the slots the checks service has spare. Each judged
2058	2060	* batch is kept for the red-batch rate (#6427); its gates and red marks are
2059	2061	* kept in the database, and a gate a roll left running adopted (#6611).
2060	2062	*/
2061	2063	function queueOptions(app: App, repo: Repo): QueueOptions {
2062	2064	return {
2063	2065	depth: queueDepth(app.db),
	2066	batch: queueBatch(app.db),
2064	2067	room: () => checksFor(app)?.spareSlots?.() ?? 0,
2065	2068	onBatch: (repoId, red) => keepBatch(app.db, repoId, red),
2066	2069	store: queueStore(app.db),
		@@ -0,0 +1,148 @@
	1	/**
	2	* The batch size (#6674): `queue.batch` caps how many pulls a candidate
	3	* carries; a red batch halves the line's size and two greens in a row
	4	* double it back; and a red batch of 16 with one bad pull lands its 15 good
	5	* in two rounds — √16 − 1 prefixes, then one gate per pull of the red run,
	6	* each leaving that pull out. The gate is held open and let go a wave at a
	7	* time, so "at once" is counted, never timed; it reads the candidate's tree:
	8	* red when it holds `bad.txt`.
	9	*/
	10	import { afterAll, beforeAll, describe, expect, test } from "bun:test";
	11	import { mkdtemp, rm } from "node:fs/promises";
	12	import { tmpdir } from "node:os";
	13	import { join } from "node:path";
	14	import { batchWords, type Candidate, enqueue, type Expect, type Queued } from "../src/checks/queue.ts";
	15	import type { LandVerdict } from "../src/checks/tiers.ts";
	16	import type { Pull } from "../src/core/pulls.ts";
	17
	18	describe("the merge queue's batch size", () => {
	19	let dir: string;
	20	const told: number[] = [];
	21	let landings = 0;
	22
	23	const git = (...args: string[]) =>
	24	Bun.spawnSync(["git", "-C", dir, "-c", "user.name=t", "-c", "user.email=t@t", ...args]).stdout.toString().trim();
	25
	26	beforeAll(async () => {
	27	dir = await mkdtemp(join(tmpdir(), "forge-merge-queue-batch-"));
	28	Bun.spawnSync(["git", "init", "-q", "-b", "main", dir]);
	29	git("commit", "-q", "--allow-empty", "-m", "start");
	30	});
	31
	32	afterAll(async () => {
	33	await rm(dir, { recursive: true, force: true });
	34	});
	35
	36	async function headAdding(branch: string, file: string): Promise<string> {
	37	git("checkout", "-q", "-B", branch, "main");
	38	git("reset", "-q", "--hard", "main");
	39	await Bun.write(join(dir, file), `${file}\n`);
	40	git("add", file);
	41	git("commit", "-q", "-m", file);
	42	git("checkout", "-q", "--detach");
	43	return git("rev-parse", branch);
	44	}
	45
	46	const verdictOf = (candidate: Candidate): LandVerdict =>
	47	git("ls-tree", "--name-only", candidate.tipOid).split("\n").includes("bad.txt")
	48	? { lands: false, reopen: true, comment: "red", learn: [] }
	49	: { lands: true };
	50
	51	function entry(line: string, number: number, headOid: string): Queued {
	52	return {
	53	pull: { id: `${line}-${number}`, number, baseBranch: "main", headOid } as Pull,
	54	dir,
	55	land: async (expect: Expect) => {
	56	if (git("rev-parse", "refs/heads/main") !== expect.baseOid) return undefined;
	57	const tip = git("commit-tree", expect.treeOid, "-p", expect.baseOid, "-m", `#${number}`);
	58	git("update-ref", "refs/heads/main", tip);
	59	landings++;
	60	return tip;
	61	},
	62	release: () => {},
	63	handBack: () => told.push(number),
	64	};
	65	}
	66
	67	async function pulls(line: string, from: number, count: number, bad?: number): Promise<Queued[]> {
	68	const out: Queued[] = [];
	69	for (let n = from; n < from + count; n++)
	70	out.push(entry(line, n, await headAdding(`${line}${n}`, n === bad ? "bad.txt" : `${line}${n}.txt`)));
	71	return out;
	72	}
	73
	74	/** A gate that answers at once and keeps how many pulls each candidate carried. */
	75	function counting() {
	76	const sizes: number[] = [];
	77	const gate = async (candidate: Candidate) => {
	78	sizes.push(candidate.steps.length);
	79	return verdictOf(candidate);
	80	};
	81	return { gate, sizes };
	82	}
	83
	84	test("the setting is honoured: 4 a candidate when told 4, 16 when told 16, never past 32", async () => {
	85	const four = counting();
	86	expect(await enqueue("told-four", await pulls("f", 1, 10), four.gate, { depth: 1, batch: 4 })).toEqual(
	87	Array(10).fill(true),
	88	);
	89	expect(four.sizes).toEqual([4, 4, 2]);
	90	const sixteen = counting();
	91	await enqueue("told-sixteen", await pulls("s", 1, 20), sixteen.gate, { depth: 1, batch: 16 });
	92	expect(sixteen.sizes).toEqual([16, 4]);
	93	expect(batchWords("told-sixteen", 16)).toBe("16");
	94	expect(batchWords("told-sixteen", 99)).toBe("32");
	95	});
	96
	97	test("a red halves the batch, and two greens in a row double it back to the setting", async () => {
	98	told.length = 0;
	99	const { gate, sizes } = counting();
	100	// Eight, the first bad: red. The search finds it; the seven land.
	101	const first = await enqueue("adapts", await pulls("a", 1, 8, 1), gate, { depth: 1, batch: 8 });
	102	expect(first).toEqual([false, ...Array(7).fill(true)]);
	103	expect(told).toEqual([1]);
	104	expect(batchWords("adapts", 8)).toBe("4 of 8");
	105	// Sixteen good: two of four, both green, so the third is eight again.
	106	sizes.length = 0;
	107	expect(await enqueue("adapts", await pulls("a", 101, 16), gate, { depth: 1, batch: 8 })).toEqual(
	108	Array(16).fill(true),
	109	);
	110	expect(sizes).toEqual([4, 4, 8]);
	111	expect(batchWords("adapts", 8)).toBe("8");
	112	});
	113
	114	test("a red 16-batch with one bad pull lands its 15 good within 2 rounds, then the bad one is handed back alone", async () => {
	115	told.length = 0;
	116	landings = 0;
	117	let pending: { candidate: Candidate; go: (v: LandVerdict) => void }[] = [];
	118	const seen: number[][][] = [];
	119	const gate = (candidate: Candidate) => new Promise<LandVerdict>((go) => pending.push({ candidate, go }));
	120	const letGo = async (n: number) => {
	121	for (let i = 0; i < 2000 && pending.length < n; i++) await Bun.sleep(5);
	122	expect(pending).toHaveLength(n);
	123	const going = pending;
	124	pending = [];
	125	seen.push(going.map((p) => p.candidate.steps.map((s) => s.entry.pull.number)));
	126	for (const p of going) p.go(verdictOf(p.candidate));
	127	};
	128	const line = await pulls("r", 1, 16, 6);
	129	// No free slot: the search still gates √16 − 1 = 3 prefixes.
	130	const landed = enqueue("red-sixteen", line, gate, { depth: 1, room: () => 0, batch: 16 });
	131	await letGo(1); // the sixteen: red
		junipercrew 18d ago Polling loops with sleep are flake-prone. Await a gate-count promise resolved from the gate callback instead.
	132	await letGo(3); // round one: 4, 8 and 12 deep; the first four land, the red run is 5–8
	133	expect(seen[1]!.map((c) => c.length)).toEqual([4, 8, 12]);
	134	for (let i = 0; i < 2000 && pending.length < 4; i++) await Bun.sleep(5);
	135	expect(landings).toBe(4);
	136	await letGo(4); // round two: everything left but one of 5–8 each; the one without 6 lands
	137	expect(seen[2]!.map((c) => c.length)).toEqual([11, 11, 11, 11]);
	138	for (let i = 0; i < 2000 && landings < 15; i++) await Bun.sleep(5);
	139	expect(landings).toBe(15);
	140	expect(git("ls-tree", "--name-only", "refs/heads/main").split("\n").filter((f) => /^r\d+\.txt$/.test(f))).toHaveLength(15);
	141	await letGo(1); // the one left out, alone: red, handed back
	142	expect(seen[3]).toEqual([[6]]);
	143	expect(await landed).toEqual(line.map((q) => q.pull.number !== 6));
	144	expect(told).toEqual([6]);
	145	expect(git("ls-tree", "--name-only", "refs/heads/main").split("\n")).not.toContain("bad.txt");
	146	expect(batchWords("red-sixteen", 16)).toBe("8 of 16");
	147	});
	148	});
		@@ -11,7 +11,7 @@ import { afterAll, beforeAll, describe, expect, test } from "bun:test";
11	11	import { mkdtemp, rm } from "node:fs/promises";
12	12	import { tmpdir } from "node:os";
13	13	import { join } from "node:path";
14		import { type Candidate, enqueue, type Expect, queueClashes, type Queued, redBatchRate, redMark } from "../src/checks/queue.ts";
	14	import { batchWords, type Candidate, enqueue, type Expect, queueClashes, type Queued, redBatchRate, redMark } from "../src/checks/queue.ts";
15	15	import type { Pull } from "../src/core/pulls.ts";
16	16
17	17	describe("the merge queue remembers clashes", () => {
		@@ -177,12 +177,18 @@ describe("the merge queue remembers clashes", () => {
177	177	expect(redMark("no-such-line", "clash-2", head)).toBeUndefined();
178	178	});
179	179
180		test("a push is a new head: B is forgotten, batched with the rest, and lands", async () => {
	180	test("a push is a new head: B is forgotten, gated as a fresh pull, and lands", async () => {
181	181	const pushed = entry(2, await headAdding("b2", "b2.txt"));
182	182	const g = entry(7, await headAdding("g", "g.txt"));
183	183	const before = gates.length;
184	184	expect(await enqueue(repoId, [pushed, g], gate)).toEqual([true, true]);
185		expect(gates.slice(before)).toEqual([{ pulls: [2, 7], ok: true }]);
	185	// Not a re-check: B goes first, as fresh. Its red batches halved this
	186	// line's batch to one (#6674); two greens in a row double it.
	187	expect(gates.slice(before)).toEqual([
	188	{ pulls: [2], ok: true },
	189	{ pulls: [7], ok: true },
	190	]);
	191	expect(batchWords(repoId, 8)).toBe("2 of 8");
186	192	expect(queueClashes(repoId)).toBe(0);
187	193	});
188	194	});
		@@ -13,8 +13,9 @@ 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, gateDeadlineForTest, MAX_CANDIDATE, gateRuns, queueStatus, queueWords, type Queued } from "../src/checks/queue.ts";
	16	import { batchWords, 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	import { queueBatch, setCapacity, validCapacity } from "../src/core/capacity.ts";
18	19	import { latestDecision } from "../src/core/decisions.ts";
19	20	import { getPull, type Pull } from "../src/core/pulls.ts";
20	21	import { getRepoById, type Repo } from "../src/core/repos.ts";
		@@ -225,6 +226,31 @@ describe("the merge queue", () => {
225	226	expect(queue.some((l) => l.includes("[queue] gated 2 pulls: green"))).toBe(true);
226	227	}, SLOW);
227	228
	229	test("queue.batch is read on each sweep: three per candidate when it says 3, and /settings/capacity explains the trade-off (#6674)", async () => {
	230	const repo = await autonomousRepo("batch-of-three");
	231	const base = git(repo, "rev-parse", "refs/heads/main");
	232	const pulls = [];
	233	for (let i = 0; i < 7; i++) pulls.push(await greenPull(repo, `b${i}.ts`));
	234	expect(validCapacity("queue.batch", "33")).toBe(false);
	235	expect(validCapacity("queue.batch", "0")).toBe(false);
	236	setCapacity(app.db, "queue.batch", "3", alice.id);
	237	try {
	238	expect(await sweepGreen(app, repo)).toBe(7);
	239	} finally {
	240	setCapacity(app.db, "queue.batch", null, alice.id);
	241	}
	242	const carried = gatesOf(repo)
	243	.map((g) => Number(git(repo, "rev-list", "--count", `${base}..${g.commit}`)))
	244	.sort((a, b) => a - b);
	245	expect(carried).toEqual([3, 6, 7]);
	246	for (const pull of pulls) expect(merged(repo, pull)).toBe(true);
	247	expect(batchWords(repo.id, queueBatch(app.db))).toBe("8");
	248	const page = await fetch(`${servers.url}/settings/capacity`, { headers: { cookie: await login() } });
	249	const html = await page.text();
	250	expect(html).toContain("Merge queue batch size");
	251	expect(html).toContain("a red one costs more search gates");
	252	}, SLOW);
	253
228	254	test("after a red batch, the good pulls go back into ONE candidate up to the cap, not one gate each (#4053)", async () => {
229	255	const repo = await autonomousRepo("rebatch");
230	256	const pulls = [];