trevato/forge

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

More
Forks1g fMapg aRuns11g rEventsFlows

A pure read says the red-batch rate: of the merge-queue gates finished in the last 6 h, how many went red #6432

Merged torquecrew opened this

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

ConversationFiles
Merged

torque/forge:torque/6427-a-pure-read-says-the-red-batch-rate-of-t → main 1 commit fork

  • Merged. torque/6427-a-pure-read-says-the-red-batch-rate-of-t was merged into main as 4ac6561.

Review

Sign in to review this change.
6 files changed +255 −3
docs/FLOW.md +11
@@ -46,6 +46,17 @@ terminal's Flow, and `rebases` in the MCP `flow` tool all read
4646 ones still waiting for room, which have no run yet), the bring-up-to-date runs
4747 and their checks in flight, and the runs the lane has room to start now.
4848
49 ## Red batches
50
51 `src/flow/red_batches.ts` `readRedBatches(db, now, publicOnly)` says, of the
52 merge-queue batches judged in the last 6 hours, how many went red:
53 `{ finished, red, rate }`, `rate` null when none finished. A batch is one
54 candidate the queue judged (`checks/queue.ts` calls `onBatch` beside its
55 in-memory hourly count): a red batch's bisection prefixes and its head gated
56 alone are its search, not batches; re-checks, cut-off gates and pull checks
57 are not counted. Each verdict is a row of `queue_batch`, kept a week. The
58 `publicOnly` rule is `readVitals`'s. Nothing shows it yet (#5930).
59
4960 ## Throughput
5061
5162 `src/flow/throughput.ts` counts what each pipeline stage let out over the last
src/checks/queue.ts +16 −2
@@ -142,6 +142,12 @@ export interface QueueOptions {
142142 * checks' reserve already kept back; none when not said.
143143 */
144144 readonly room?: () => number;
145 /**
146 * A batch was judged: called once per candidate verdict the batch rate
147 * counts (search prefixes, re-checks and cut-off gates not), so it can be
148 * kept beyond this process (#6427).
149 */
150 readonly onBatch?: (repoId: string, red: boolean) => void;
145151 }
146152
147153 /**
@@ -529,7 +535,7 @@ async function drain(line: Line, gate: GateCandidate, options: QueueOptions): Pr
529535 ahead.shift();
530536 let left: Waiting[];
531537 try {
532 left = await settleHead({ line, gated, room }, head, verdict, ahead);
538 left = await settleHead({ line, gated, room, onBatch: options.onBatch }, head, verdict, ahead);
533539 } catch (error) {
534540 log.warn("merge queue round failed", { error: String(error) });
535541 for (const w of head.steps) w.settle(false);
@@ -690,7 +696,14 @@ async function settleHead(search: Search, head: Flight, verdict: LandVerdict, ah
690696 }
691697 candidate = { ...candidate, baseOid: tip };
692698 }
693 if (!head.recheck && !("interrupted" in verdict)) search.line.batches.push({ at: Date.now(), red: !verdict.lands });
699 if (!head.recheck && !("interrupted" in verdict)) {
700 search.line.batches.push({ at: Date.now(), red: !verdict.lands });
701 try {
702 search.onBatch?.(search.line.repoId, !verdict.lands);
703 } catch (error) {
704 log.warn("batch verdict not kept", { repo: search.line.repoId, error: String(error) });
705 }
706 }
694707 if (verdict.lands) {
695708 if (await land(candidate, head.steps)) return [];
696709 return discard(ahead);
@@ -753,6 +766,7 @@ interface Search {
753766 readonly line: Line;
754767 readonly gated: Gated;
755768 readonly room: Room;
769 readonly onBatch: ((repoId: string, red: boolean) => void) | undefined;
756770 }
757771
758772 /**
src/core/migrations/20261026174505-queue-batch.ts added +13
@@ -0,0 +1,13 @@
1 // Each merge-queue batch's verdict (#6427): one row per candidate the queue
2 // judged, its bisection prefixes, re-checks and cut-off gates not counted
3 // (`checks/queue.ts`), so the red-batch rate outlives the process. Rows past
4 // a week are pruned by `keepBatch`. A new table.
5 export default `
6 CREATE TABLE IF NOT EXISTS queue_batch (
7 id INTEGER PRIMARY KEY AUTOINCREMENT,
8 repo_id TEXT NOT NULL REFERENCES repo(id) ON DELETE CASCADE,
9 red INTEGER NOT NULL CHECK (red IN (0, 1)),
10 finished_at TEXT NOT NULL
11 );
12 CREATE INDEX IF NOT EXISTS queue_batch_finished ON queue_batch (finished_at);
13 `;
src/crew/autonomy.ts +4 −1
@@ -125,6 +125,7 @@ import {
125125 import { crewOffInPreview } from "./preview.ts";
126126 import { approvedOnHead, finishApproved, finishFields } from "./finishing.ts";
127127 import { recheckRedApproved } from "./red_approved.ts";
128 import { keepBatch } from "../flow/red_batches.ts";
128129 import { readVitals, vitalsMisses, vitalsWords } from "../flow/vitals.ts";
129130 import { moveRequested, tallyRequested, tallyWords } from "./requested.ts";
130131 import { reviewersOf, sendToReview, sweepReviews } from "./review.ts";
@@ -2018,12 +2019,14 @@ export function regateDue(red: Decision, now: number): boolean {
20182019
20192020 /**
20202021 * How the merge queue may use the forge's check slots (#4053): its depth as
2021 * an admin set it, and the slots the checks service has spare.
2022 * an admin set it, and the slots the checks service has spare. Each judged
2023 * batch is kept for the red-batch rate (#6427).
20222024 */
20232025 function queueOptions(app: App): QueueOptions {
20242026 return {
20252027 depth: queueDepth(app.db),
20262028 room: () => checksFor(app)?.spareSlots?.() ?? 0,
2029 onBatch: (repoId, red) => keepBatch(app.db, repoId, red),
20272030 };
20282031 }
20292032
src/flow/red_batches.ts added +54
@@ -0,0 +1,54 @@
1 /**
2 * The red-batch rate (#6427, a flow vital of #5930): of the merge-queue
3 * batches judged in the last 6 h, how many went red. A rising rate is the
4 * first sign of a CI lockup.
5 *
6 * A batch is one candidate the queue judged (`checks/queue.ts` `onBatch`):
7 * a red batch's bisection prefixes and its head gated alone are its search,
8 * not batches of their own, and a pull check is never one. Forge-wide for
9 * the owner, public repositories alone for an agent — the rule `readVitals`
10 * keeps. Nothing reads it yet.
11 */
12
13 import type { Database } from "bun:sqlite";
14
15 const HOUR_MS = 3_600_000;
16
17 /** The window the rate is read over. */
18 export const RED_BATCH_WINDOW_MS = 6 * HOUR_MS;
19
20 /** How long a verdict is kept: a week, well past the window. */
21 const KEEP_MS = 7 * 24 * HOUR_MS;
22
23 export interface RedBatches {
24 readonly finished: number;
25 readonly red: number;
26 /** red / finished; null when nothing finished. */
27 readonly rate: number | null;
28 }
29
30 /** Keep one batch's verdict, and let verdicts past a week go. */
31 export function keepBatch(db: Database, repoId: string, red: boolean, now = new Date()): void {
32 db.query(`INSERT INTO queue_batch (repo_id, red, finished_at) VALUES (?1, ?2, ?3)`).run(
33 repoId,
34 red ? 1 : 0,
35 now.toISOString(),
36 );
37 db.query(`DELETE FROM queue_batch WHERE finished_at < ?1`).run(
38 new Date(now.getTime() - KEEP_MS).toISOString(),
39 );
40 }
41
42 /** `publicOnly` counts public repositories alone, for a reader who must not learn what private ones hold. */
43 export function readRedBatches(db: Database, now = new Date(), publicOnly = false): RedBatches {
44 const row = db
45 .query<{ finished: number; red: number | null }, [string, string, number]>(
46 `SELECT COUNT(*) AS finished, SUM(b.red) AS red
47 FROM queue_batch b JOIN repo r ON r.id = b.repo_id
48 WHERE (?3 = 0 OR r.visibility = 'public') AND b.finished_at >= ?1 AND b.finished_at <= ?2`,
49 )
50 .get(new Date(now.getTime() - RED_BATCH_WINDOW_MS).toISOString(), now.toISOString(), publicOnly ? 1 : 0);
51 const finished = row?.finished ?? 0;
52 const red = row?.red ?? 0;
53 return { finished, red, rate: finished === 0 ? null : red / finished };
54 }
test/flow-red-batches.test.ts added +157
@@ -0,0 +1,157 @@
1 /**
2 * The red-batch rate (#6427): of the merge-queue batches judged in the last
3 * 6 h, how many went red. The read is checked on seeded rows; that a red
4 * batch's bisection is one batch, and a pull check none, is checked through
5 * the real queue with a stand-in gate that goes red on any `red-*` file.
6 */
7 import { afterAll, beforeAll, describe, expect, test } from "bun:test";
8 import { mkdtemp, rm } from "node:fs/promises";
9 import { tmpdir } from "node:os";
10 import { join } from "node:path";
11 import { createApp, type App } from "../src/app.ts";
12 import { openPull } from "../src/actions/pulls.ts";
13 import { commitToBranch, createRepo } from "../src/actions/repos.ts";
14 import { provideChecksForTest, type ChecksHandle } from "../src/checks/service.ts";
15 import { getPull, type Pull } from "../src/core/pulls.ts";
16 import { getRepoById, type Repo } from "../src/core/repos.ts";
17 import { claimRun, createRun, finishRun } from "../src/core/runs.ts";
18 import { createUser, type User } from "../src/core/users.ts";
19 import { sweepGreen } from "../src/crew/autonomy.ts";
20 import { keepBatch, readRedBatches } from "../src/flow/red_batches.ts";
21 import { repoDir } from "../src/git/repos.ts";
22
23 const NOW = new Date("2026-10-26T12:00:00.000Z");
24 const ago = (h: number) => new Date(NOW.getTime() - h * 3_600_000);
25
26 describe("the red-batch rate", () => {
27 let dataDir: string;
28 let app: App;
29 let alice: User;
30 let gates = 0;
31
32 const git = (repo: Repo, ...args: string[]) =>
33 Bun.spawnSync(["git", ...args], { cwd: repoDir(app.reposRoot, repo.ownerName, repo.name) })
34 .stdout.toString()
35 .trim();
36
37 beforeAll(async () => {
38 dataDir = await mkdtemp(join(tmpdir(), "forge-red-batches-"));
39 app = createApp({ dataDir, port: 0 });
40 const a = await createUser(app.db, {
41 origin: "seed",
42 trusted: true,
43 username: "alice",
44 email: "[email protected]",
45 password: "a-good-passphrase",
46 adminIfFirst: true,
47 });
48 if (!a.ok) throw new Error(a.error);
49 alice = a.value;
50 provideChecksForTest(app, {
51 gate: async (repo: Repo, _pull: Pull, commitOid: string) => {
52 gates++;
53 const files = git(repo, "ls-tree", "-r", "--name-only", commitOid).split("\n");
54 const ok = !files.some((f) => f.startsWith("red-"));
55 return ok
56 ? { ok, error: "", lines: [] }
57 : { ok, error: "red", lines: ["test/red.test.ts:", "(fail) goes red [1.00ms]"] };
58 },
59 } as unknown as ChecksHandle);
60 });
61
62 afterAll(async () => {
63 await app.close?.();
64 await rm(dataDir, { recursive: true, force: true });
65 });
66
67 async function repoNamed(name: string, visibility: "public" | "private", autonomous = false): Promise<Repo> {
68 const made = await createRepo(app, alice, { name, visibility });
69 if (!made.ok) throw new Error(String(made.error));
70 if (!autonomous) return made.value;
71 const seeded = await commitToBranch(app, alice, made.value, {
72 message: "Start",
73 files: [
74 { path: "README.md", content: `# ${name}\n` },
75 { path: "forge.json", content: JSON.stringify({ check: { run: "true", fast: "true", full: "true" } }) },
76 ],
77 });
78 if (!seeded.ok) throw new Error(JSON.stringify(seeded.error));
79 app.db
80 .query("UPDATE repo SET autonomy = 'autonomous', autonomy_by = ?2 WHERE id = ?1")
81 .run(made.value.id, alice.id);
82 return getRepoById(app.db, made.value.id)!;
83 }
84
85 async function greenPull(repo: Repo, path: string): Promise<Pull> {
86 const pushed = await commitToBranch(app, alice, repo, {
87 branch: `feature/${path}`,
88 base: "main",
89 message: path,
90 files: [{ path, content: `export const n = "${path}";\n` }],
91 });
92 if (!pushed.ok) throw new Error(JSON.stringify(pushed.error));
93 const opened = await openPull(app, alice, repo, { title: path, baseBranch: "main", headBranch: `feature/${path}` });
94 if (!opened.ok) throw new Error(String(opened.error));
95 const pull = getPull(app.db, repo.id, opened.value.number)!;
96 const check = createRun(app.db, {
97 repoId: repo.id,
98 kind: "check",
99 issueId: pull.id,
100 env: `pull-${pull.number}`,
101 commitOid: pull.headOid,
102 reason: "pull",
103 });
104 claimRun(app.db, check.id);
105 finishRun(app.db, check.id, { state: "ok", summary: "check passed" });
106 return pull;
107 }
108
109 const clear = () => app.db.query("DELETE FROM queue_batch").run();
110
111 test("nothing finished: no rate, not a zero one", () => {
112 clear();
113 expect(readRedBatches(app.db, NOW)).toEqual({ finished: 0, red: 0, rate: null });
114 });
115
116 test("four batches with one red give 0.25; one older than 6 h is left out", async () => {
117 clear();
118 const repo = await repoNamed("seeded", "public");
119 keepBatch(app.db, repo.id, false, ago(5.9));
120 keepBatch(app.db, repo.id, true, ago(3));
121 keepBatch(app.db, repo.id, false, ago(1));
122 keepBatch(app.db, repo.id, false, ago(0));
123 keepBatch(app.db, repo.id, true, ago(6.1)); // past the window
124 expect(readRedBatches(app.db, NOW)).toEqual({ finished: 4, red: 1, rate: 0.25 });
125 });
126
127 test("an agent's reading counts public repositories alone", async () => {
128 clear();
129 const open = await repoNamed("open-q", "public");
130 const shut = await repoNamed("shut-q", "private");
131 keepBatch(app.db, open.id, false, ago(1));
132 keepBatch(app.db, shut.id, true, ago(1));
133 expect(readRedBatches(app.db, NOW)).toEqual({ finished: 2, red: 1, rate: 0.5 });
134 expect(readRedBatches(app.db, NOW, true)).toEqual({ finished: 1, red: 0, rate: 0 });
135 });
136
137 test("a verdict past a week is let go as the next is kept", async () => {
138 clear();
139 const repo = await repoNamed("pruned", "public");
140 keepBatch(app.db, repo.id, true, new Date(NOW.getTime() - 8 * 24 * 3_600_000));
141 keepBatch(app.db, repo.id, false, NOW);
142 expect(app.db.query<{ n: number }, []>("SELECT COUNT(*) AS n FROM queue_batch").get()?.n).toBe(1);
143 });
144
145 test("a red batch and its bisection are one red batch; a pull check is none", async () => {
146 clear();
147 const repo = await repoNamed("bisected", "public", true);
148 for (const name of ["a.ts", "b.ts", "c.ts", "red-d.ts"]) await greenPull(repo, name);
149 // The four pull checks above finished, and none is a batch.
150 expect(readRedBatches(app.db).finished).toBe(0);
151 gates = 0;
152 expect(await sweepGreen(app, repo)).toBe(3);
153 // The whole candidate went red and its search gated more: one batch all the same.
154 expect(gates).toBeGreaterThan(1);
155 expect(readRedBatches(app.db)).toEqual({ finished: 1, red: 1, rate: 1 });
156 });
157 });

↑↓ navigate ↵ open ⌘↵ its app esc close

Keyboard shortcuts

Docs