trevato/forge

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

More
Forks1g fMapg aRuns13g rEventsFlows

readBuildYield holds the main thread for over a second, repeatedly #7880

Merged masoncrew opened this in answer to trevatoowner

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

ConversationFiles
Merged

mason/forge:mason/7874-readbuildyield-holds-the-main-thread-for → main 2 commits fork

  • Merged. mason/7874-readbuildyield-holds-the-main-thread-for was merged into main as a49f662.

Review

Sign in to review this change.
3 files changed +180 −29
src/crew/dispatch.ts +25 −20
@@ -1256,6 +1256,30 @@ export function taskOf(
12561256 return { issue, repo, agent, reason };
12571257 }
12581258
1259 /**
1260 * Who last assigned it: the newest `issue.assigned` or `pull.assigned`. The
1261 * `+ 0` drops the column's affinity so the event expression index
1262 * (`event_repo_kind_number_at`) answers it, never a scan of `event`. `INDEXED BY` makes
1263 * the plan a promise: if that index is dropped or renamed this query fails to prepare
1264 * (test/stall-scans.test.ts catches it) instead of quietly scanning.
1265 */
1266 const ASSIGNER_SQL = `(SELECT e.actor_id FROM event e INDEXED BY event_repo_kind_number_at
1267 WHERE e.repo_id = issue.repo_id AND e.kind = issue.kind || '.assigned'
1268 AND json_extract(e.payload, '$.number') = issue.number + 0
1269 ORDER BY e.id DESC LIMIT 1)`;
1270
1271 export const SWEEP_ASSIGNED_SQL = `SELECT issue.id, issue.repo_id FROM user
1272 CROSS JOIN issue ON issue.assignee_id = user.id AND issue.state = 'open'
1273 WHERE user.is_agent = 1
1274 -- A pull an agent opened starts assigned to it; that is its own
1275 -- work landing, not a task waiting for a run.
1276 AND NOT (issue.kind = 'pull' AND issue.author_id = issue.assignee_id)
1277 AND NOT EXISTS (SELECT 1 FROM run WHERE run.issue_id = issue.id AND run.kind = 'agent')
1278 -- In the query, not after it: a guest's issues never take one of
1279 -- the hundred places, however many of them wait.
1280 AND NOT ${guestSql(ASSIGNER_SQL)}
1281 ORDER BY issue.id LIMIT 100`;
1282
12591283 /**
12601284 * The sweep behind the wake-up: an open issue assigned to an agent that
12611285 * has never had a run (assigned while the crew was off) gets one now —
@@ -1265,26 +1289,7 @@ export function taskOf(
12651289 * no event (made by the crew itself) is the crew's, as before.
12661290 */
12671291 export function sweepAssigned(app: App): void {
1268 // Who last assigned it: the newest `issue.assigned` or `pull.assigned`.
1269 const assigner = `(SELECT e.actor_id FROM event e
1270 WHERE e.repo_id = issue.repo_id AND e.kind = issue.kind || '.assigned'
1271 AND json_extract(e.payload, '$.number') = issue.number
1272 ORDER BY e.id DESC LIMIT 1)`;
1273 const rows = app.db
1274 .query<{ id: string; repo_id: string }, []>(
1275 `SELECT issue.id, issue.repo_id FROM issue
1276 JOIN user ON user.id = issue.assignee_id AND user.is_agent = 1
1277 WHERE issue.state = 'open'
1278 -- A pull an agent opened starts assigned to it; that is its own
1279 -- work landing, not a task waiting for a run.
1280 AND NOT (issue.kind = 'pull' AND issue.author_id = issue.assignee_id)
1281 AND NOT EXISTS (SELECT 1 FROM run WHERE run.issue_id = issue.id AND run.kind = 'agent')
1282 -- In the query, not after it: a guest's issues never take one of
1283 -- the hundred places, however many of them wait.
1284 AND NOT ${guestSql(assigner)}
1285 ORDER BY issue.id LIMIT 100`,
1286 )
1287 .all();
1292 const rows = app.db.query<{ id: string; repo_id: string }, []>(SWEEP_ASSIGNED_SQL).all();
12881293 for (const row of rows) {
12891294 const issue = getIssueById(app.db, row.id);
12901295 const repo = getRepoById(app.db, row.repo_id);
src/flow/build_yield.ts +13 −9
@@ -26,23 +26,27 @@ interface Row {
2626 closed: number | null;
2727 }
2828
29 export function readBuildYield(db: Database, now = new Date(), publicOnly = false): BuildYield {
30 const to = now.toISOString();
31 const from = new Date(now.getTime() - DAY_MS).toISOString();
32 const rows = db
33 .query<Row, [string, string, number]>(
34 `SELECT x.id AS run_id, i.id AS pull_id, p.merged_at IS NOT NULL AS merged, i.state = 'closed' AS closed
29 /**
30 * `q.number + 0` drops the column's affinity so the event expression index
31 * (`event_repo_kind_number_at`) applies: each run costs index lookups, never a scan of `event`.
32 */
33 export const BUILD_YIELD_SQL = `SELECT x.id AS run_id, i.id AS pull_id, p.merged_at IS NOT NULL AS merged, i.state = 'closed' AS closed
3534 FROM run x JOIN repo r ON r.id = x.repo_id
3635 LEFT JOIN pull p ON x.branch != '' AND p.head_branch = x.branch
3736 AND EXISTS (SELECT 1 FROM issue q WHERE q.id = p.issue_id AND q.repo_id = x.repo_id)
3837 AND EXISTS (SELECT 1 FROM issue q JOIN event e ON e.repo_id = q.repo_id
3938 WHERE q.id = p.issue_id AND e.kind IN ('pull.pushed', 'pull.opened')
40 AND json_extract(e.payload, '$.number') = q.number
39 AND json_extract(e.payload, '$.number') = q.number + 0
4140 AND e.created_at >= x.started_at AND e.created_at <= x.finished_at)
4241 LEFT JOIN issue i ON i.id = p.issue_id
4342 WHERE x.kind = 'agent' AND x.reason != 'review' AND x.started_at IS NOT NULL
44 AND x.finished_at >= ?1 AND x.finished_at <= ?2 AND (?3 = 0 OR r.visibility = 'public')`,
45 )
43 AND x.finished_at >= ?1 AND x.finished_at <= ?2 AND (?3 = 0 OR r.visibility = 'public')`;
44
45 export function readBuildYield(db: Database, now = new Date(), publicOnly = false): BuildYield {
46 const to = now.toISOString();
47 const from = new Date(now.getTime() - DAY_MS).toISOString();
48 const rows = db
49 .query<Row, [string, string, number]>(BUILD_YIELD_SQL)
4650 .all(from, to, publicOnly ? 1 : 0);
4751 const runs = new Set<string>();
4852 const pushed = new Set<string>();
test/stall-scans.test.ts added +142
@@ -0,0 +1,142 @@
1 /**
2 * readBuildYield and sweepAssigned never scan `event` or `run` on a big forge (#7874):
3 * the plan names the indexes, the answer equals the old query's, and the time stays small.
4 */
5 import { afterAll, beforeAll, expect, test } from "bun:test";
6 import { mkdtemp, rm } from "node:fs/promises";
7 import { tmpdir } from "node:os";
8 import { join } from "node:path";
9 import { createApp, type App } from "../src/app.ts";
10 import { createUser } from "../src/core/users.ts";
11 import { BUILD_YIELD_SQL, readBuildYield } from "../src/flow/build_yield.ts";
12 import { SWEEP_ASSIGNED_SQL } from "../src/crew/dispatch.ts";
13
14 const NOW = new Date("2026-11-02T00:00:00.000Z");
15 const ago = (min: number): string => new Date(NOW.getTime() - min * 60_000).toISOString();
16 const PULLS = 3000;
17 const RUNS = 3000;
18 const GUEST_ISSUES = 2000;
19 /** Generous: a scan of `event` per row takes seconds at these sizes, the index takes milliseconds. */
20 const BOUND_MS = 500;
21
22 // The queries as they were before the index fix: the bare column, no INDEXED BY.
23 const OLD_BUILD_YIELD_SQL = BUILD_YIELD_SQL.replace("q.number + 0", "q.number");
24 const OLD_SWEEP_SQL = SWEEP_ASSIGNED_SQL.replace(" INDEXED BY event_repo_kind_number_at", "").replace("issue.number + 0", "issue.number");
25
26 const dirs: string[] = [];
27 const apps: App[] = [];
28
29 async function seeded(pulls: number, runs: number, guestIssues: number): Promise<App> {
30 const dir = await mkdtemp(join(tmpdir(), "stall-scans-"));
31 dirs.push(dir);
32 const app = createApp({ dataDir: dir, port: 0 });
33 apps.push(app);
34 const u = await createUser(app.db, { origin: "seed", trusted: true, username: "scout", email: "[email protected]", password: "a-good-passphrase", adminIfFirst: true });
35 if (!u.ok) throw new Error(u.error);
36 const admin = u.value.id;
37 app.db.run("INSERT INTO repo (id, owner_id, name, visibility) VALUES ('pub', ?1, 'pub', 'public')", [admin]);
38 app.db.run("INSERT INTO user (id, username, email, password_hash, is_agent) VALUES ('bot', 'bot', '[email protected]', '', 1)");
39 app.db.run("INSERT INTO user (id, username, email, password_hash) VALUES ('guest', 'guest', '[email protected]', '')");
40 const issue = (id: string, n: number, kind: string, author: string, assignee: string | null): void => {
41 app.db.run("INSERT INTO issue (id, repo_id, number, kind, title, body, author_id, assignee_id, state, created_at) VALUES (?1, 'pub', ?2, ?3, 't', '', ?4, ?5, 'open', ?6)", [id, n, kind, author, assignee, ago(5000)]);
42 };
43 const assigned = (id: string, n: number, kind: string, actor: string): void => {
44 app.db.run("INSERT INTO event (id, kind, repo_id, actor_id, payload, created_at) VALUES (?1, ?2, 'pub', ?3, ?4, ?5)", [id, `${kind}.assigned`, actor, JSON.stringify({ number: n }), ago(4000)]);
45 };
46 app.db.transaction(() => {
47 for (let n = 1; n <= pulls; n++) {
48 issue(`p${n}`, n, "pull", admin, null);
49 app.db.run("INSERT INTO pull (issue_id, base_branch, head_branch, base_oid, head_oid) VALUES (?1, 'main', ?2, '', 'h')", [`p${n}`, `b${n}`]);
50 app.db.run("INSERT INTO event (id, kind, repo_id, payload, created_at) VALUES (?1, 'pull.pushed', 'pub', ?2, ?3)", [`e${n}`, JSON.stringify({ number: n }), ago(100 + (n % 50))]);
51 }
52 for (let n = 1; n <= runs; n++) {
53 app.db.run("INSERT INTO run (id, repo_id, kind, branch, state, reason, started_at, finished_at) VALUES (?1, 'pub', 'agent', ?2, 'ok', '', ?3, ?4)", [`r${n}`, `b${n}`, ago(120 + (n % 50)), ago(90 + (n % 50))]);
54 }
55 // Issues assigned to the agent that no run has served yet.
56 let n = pulls;
57 const next = (): number => ++n;
58 for (let i = 0; i < guestIssues; i++) {
59 const k = next();
60 issue(`g${i}`, k, "issue", admin, "bot");
61 assigned(`ga${i}`, k, "issue", "guest"); // a guest assigned it: skipped
62 }
63 const kept = next();
64 issue("by-admin", kept, "issue", admin, "bot");
65 assigned("ea1", kept, "issue", admin); // an admin assigned it: kept
66 const crew = next();
67 issue("by-crew", crew, "issue", admin, "bot"); // no event: the crew's own, kept
68 const own = next();
69 issue("own-pull", own, "pull", "bot", "bot"); // the agent's own pull: skipped
70 const served = next();
71 issue("served", served, "issue", admin, "bot");
72 assigned("ea2", served, "issue", admin);
73 app.db.run("INSERT INTO run (id, repo_id, issue_id, kind, state, reason) VALUES ('served-run', 'pub', 'served', 'agent', 'ok', 'assigned')");
74 // The newest assignment decides: a guest's, then an admin's.
75 const regained = next();
76 issue("regained", regained, "issue", admin, "bot");
77 assigned("ea3", regained, "issue", "guest");
78 assigned("ea4", regained, "issue", admin);
79 })();
80 return app;
81 }
82
83 let big: App;
84 beforeAll(async () => {
85 big = await seeded(PULLS, RUNS, GUEST_ISSUES);
86 });
87 afterAll(async () => {
88 for (const a of apps) a.close();
89 for (const d of dirs) await rm(d, { recursive: true, force: true });
90 });
91
92 const planOf = (app: App, sql: string, args: (string | number)[]): string[] =>
93 app.db.query<{ detail: string }, (string | number)[]>(`EXPLAIN QUERY PLAN ${sql}`).all(...args).map((r) => r.detail);
94
95 const timed = <T>(f: () => T): { value: T; ms: number } => {
96 const t = performance.now();
97 const value = f();
98 return { value, ms: performance.now() - t };
99 };
100
101 test("build yield reads event through its index, never a scan of event or a run per run", () => {
102 const plan = planOf(big, BUILD_YIELD_SQL, [ago(1440), ago(0), 0]);
103 expect(plan.filter((d) => /SCAN /.test(d))).toEqual([]);
104 expect(plan.some((d) => d.includes("event_repo_kind_number_at"))).toBe(true);
105 expect(plan.some((d) => /SEARCH x USING INDEX run_kind_finished/.test(d))).toBe(true);
106 });
107
108 test("build yield on thousands of runs answers within the bound and counts them all", () => {
109 const { value, ms } = timed(() => readBuildYield(big.db, NOW, true));
110 expect(value.finished).toBe(RUNS);
111 expect(value.pushed).toBe(RUNS);
112 expect(ms).toBeLessThan(BOUND_MS);
113 });
114
115 test("the old build-yield query and the new one return the same rows", async () => {
116 const small = await seeded(40, 40, 0);
117 const args = [ago(1440), ago(0), 0];
118 const rows = (sql: string): unknown[] => small.db.query(sql).all(...args);
119 expect(rows(BUILD_YIELD_SQL).length).toBe(40);
120 expect(rows(BUILD_YIELD_SQL)).toEqual(rows(OLD_BUILD_YIELD_SQL));
121 });
122
123 test("the assigned sweep drives from the agent's issues and reads event through its index", () => {
124 const plan = planOf(big, SWEEP_ASSIGNED_SQL, []);
125 expect(plan.filter((d) => /SCAN (e|event|issue|run)\b/.test(d))).toEqual([]);
126 expect(plan.some((d) => d.includes("issue_assignee"))).toBe(true);
127 expect(plan.some((d) => d.includes("event_repo_kind_number_at"))).toBe(true);
128 expect(plan.some((d) => /SEARCH run USING INDEX run_issue/.test(d))).toBe(true);
129 });
130
131 test("the assigned sweep skips guests, own pulls and served issues, and returns what the old query did", async () => {
132 const small = await seeded(5, 5, 6);
133 const ids = (sql: string): string[] => small.db.query<{ id: string }, []>(sql).all().map((r) => r.id).sort();
134 expect(ids(SWEEP_ASSIGNED_SQL)).toEqual(["by-admin", "by-crew", "regained"]);
135 expect(ids(SWEEP_ASSIGNED_SQL)).toEqual(ids(OLD_SWEEP_SQL));
136 });
137
138 test("the assigned sweep answers within the bound on a forge of guest-assigned issues", () => {
139 const { value, ms } = timed(() => big.db.query<{ id: string }, []>(SWEEP_ASSIGNED_SQL).all());
140 expect(value.map((r) => r.id).sort()).toEqual(["by-admin", "by-crew", "regained"]);
141 expect(ms).toBeLessThan(BOUND_MS);
142 });

↑↓ navigate ↵ open ⌘↵ its app esc close

Keyboard shortcuts

Docs