diff --git a/cloudflare/src/jobs/store.ts b/cloudflare/src/jobs/store.ts index 57291a43..cbe2a4be 100644 --- a/cloudflare/src/jobs/store.ts +++ b/cloudflare/src/jobs/store.ts @@ -727,12 +727,16 @@ export async function readJob(db: D1Database, id: string) { ]); const row = job!.results[0] as Job | undefined; if (!row) return null; + return jobView(row, events!.results as Event[], questions!.results as Question[]); +} + +function jobView(row: Job, events: Event[], questions: Question[]) { const json = (value: string | null): unknown => (value === null ? null : JSON.parse(value)); return { ...row, result: json(row.result), - events: (events!.results as Event[]).map((e) => ({ ...e, body: json(e.body as string) })), - questions: (questions!.results as Question[]).map((q) => ({ + events: events.map((e) => ({ ...e, body: json(e.body as string) })), + questions: questions.map((q) => ({ ...q, prompt: json(q.prompt as string), answer: json(q.answer as string | null), @@ -740,12 +744,40 @@ export async function readJob(db: D1Database, id: string) { }; } +/** Hydrate a selection in three snapshot queries, without multiplying events by questions. */ +async function readJobs(db: D1Database, select: string, ...parameters: (string | number)[]) { + const [jobs, events, questions] = await db.batch([ + db.prepare(select).bind(...parameters), + db + .prepare( + `WITH selected AS (${select}) SELECT e.job,e.attempt,e.kind,e.body,e.at + FROM job_events e JOIN selected j ON j.id=e.job ORDER BY e.rowid`, + ) + .bind(...parameters), + db + .prepare( + `WITH selected AS (${select}) SELECT q.job,q.id,q.attempt,q.kind,q.prompt,q.answer,q.answeredBy,q.askedAt,q.answeredAt,q.approvedAt + FROM job_questions q JOIN selected j ON j.id=q.job ORDER BY q.askedAt,q.rowid`, + ) + .bind(...parameters), + ]); + const byJob = (rows: T[]) => { + const grouped = new Map[]>(); + for (const { job, ...row } of rows) { + if (!grouped.has(job)) grouped.set(job, []); + grouped.get(job)!.push(row); + } + return grouped; + }; + const eventRows = byJob(events!.results as (Event & { job: string })[]); + const questionRows = byJob(questions!.results as (Question & { job: string })[]); + return (jobs!.results as Job[]).map((row) => + jobView(row, eventRows.get(row.id) ?? [], questionRows.get(row.id) ?? []), + ); +} + export async function listJobs(db: D1Database, repository: number) { - const rows = await db - .prepare("SELECT id FROM jobs WHERE repository=? ORDER BY createdAt DESC,id LIMIT 50") - .bind(repository) - .all<{ id: string }>(); - return (await Promise.all(rows.results.map((row) => readJob(db, row.id)))).filter((job) => job !== null); + return readJobs(db, "SELECT * FROM jobs WHERE repository=? ORDER BY createdAt DESC,id LIMIT 50", repository); } export type Published = Pick; @@ -777,16 +809,15 @@ export async function active(db: D1Database, repository: number, target: string) * one of the `observed` issues, pull requests, or ready quests. */ export async function listInboxJobs(db: D1Database, repository: number, observed: string[]) { - const rows = await db - .prepare( - `SELECT id FROM jobs j WHERE repository=?1 + return readJobs( + db, + `SELECT * FROM jobs j WHERE repository=?1 AND NOT EXISTS(SELECT 1 FROM jobs n WHERE n.repository=?1 AND n.target=j.target AND (n.createdAt>j.createdAt OR (n.createdAt=j.createdAt AND n.id(); - return (await Promise.all(rows.results.map((row) => readJob(db, row.id)))).filter((job) => job !== null); + repository, + JSON.stringify(observed), + ); } /** The most follow-ups a delivery suggests, and the longest one, in characters. */ @@ -848,17 +879,16 @@ export async function followupRequest(db: D1Database, repository: number, job: s * delivered is one of the `open` ones and no later job works on that pull request. */ export async function listActivityJobs(db: D1Database, repository: number, open: number[]) { - const rows = await db - .prepare( - `SELECT id FROM jobs j WHERE repository=?1 + return readJobs( + db, + `SELECT * FROM jobs j WHERE repository=?1 AND NOT EXISTS(SELECT 1 FROM jobs n WHERE n.repository=?1 AND n.target=j.target AND (n.createdAt>j.createdAt OR (n.createdAt=j.createdAt AND n.idj.id AND n.createdAt>=j.createdAt))) ORDER BY createdAt DESC,id`, - ) - .bind(repository, JSON.stringify(open)) - .all<{ id: string }>(); - return (await Promise.all(rows.results.map((row) => readJob(db, row.id)))).filter((job) => job !== null); + repository, + JSON.stringify(open), + ); } type Pin = { runner: string; account: string }; diff --git a/cloudflare/test/inbox-reads.test.ts b/cloudflare/test/inbox-reads.test.ts new file mode 100644 index 00000000..2393cddf --- /dev/null +++ b/cloudflare/test/inbox-reads.test.ts @@ -0,0 +1,155 @@ +import { env } from "cloudflare:workers"; +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; +import { load } from "../src/inbox/model"; +import * as store from "../src/jobs/store"; +import { commit, maintainerOf, project } from "./project-fixtures"; + +const db = env.DB; +let user = ""; +beforeEach(async () => { + ({ user } = await maintainerOf()); +}); +afterEach(() => vi.restoreAllMocks()); + +/** Count executed statements, including each statement inside a D1 batch. */ +function counted() { + let queries = 0; + const originals = new WeakMap(); + const statement = (inner: D1PreparedStatement): D1PreparedStatement => { + const proxy = new Proxy(inner, { + get(target, key) { + if (key === "bind") return (...args: unknown[]) => statement(target.bind(...args)); + const value = Reflect.get(target, key, target); + if (typeof value !== "function") return value; + return (...args: unknown[]) => { + if (["all", "first", "run", "raw"].includes(String(key))) queries++; + return value.apply(target, args); + }; + }, + }); + originals.set(proxy, inner); + return proxy; + }; + return { + db: new Proxy(db, { + get(target, key) { + if (key === "prepare") return (sql: string) => statement(target.prepare(sql)); + if (key === "batch") + return (statements: D1PreparedStatement[]) => { + queries += statements.length; + return target.batch(statements.map((s) => originals.get(s) ?? s)); + }; + const value = Reflect.get(target, key, target); + return typeof value === "function" ? value.bind(target) : value; + }, + }), + queries: () => queries, + }; +} + +/** Many targets, with interleaved attempts, events, answered and unanswered questions. */ +async function seed(count: number, repository = 10) { + const ids: string[] = []; + const statements: D1PreparedStatement[] = []; + for (let i = 0; i < count; i++) { + const id = `job-${repository}-${String(i).padStart(3, "0")}`; + ids.push(id); + statements.push( + db + .prepare( + `INSERT INTO jobs(id,repository,workflow,target,revision,branch,actor,trusted,state,attempt,result,createdAt,updatedAt) + VALUES(?,?,'start',?,?,'quest/work',?,1,?,2,?,?,?)`, + ) + .bind( + id, + repository, + `quest/job-${i}.md`, + commit, + user, + i % 2 ? "completed" : "waiting", + JSON.stringify({ followups: [`Follow-up ${i}`] }), + i, + 100 + i, + ), + ); + if (i % 3 === 0) continue; + // Event order follows insertion, not timestamp or attempt. + for (const attempt of [2, 1]) + statements.push( + db + .prepare("INSERT INTO job_events(job,attempt,kind,body,at) VALUES(?,?,'summary',?,?)") + .bind(id, attempt, JSON.stringify({ summary: `${id}-${attempt}` }), attempt), + ); + // Questions order by askedAt, using rowid to break ties, not by attempt. + for (const attempt of [2, 1]) + statements.push( + db + .prepare( + `INSERT INTO job_questions(id,job,attempt,kind,prompt,answer,answeredBy,askedAt,answeredAt,approvedAt) + VALUES(?,?,?,'agent',?,?,?,?,?,?)`, + ) + .bind( + `${id}-${attempt}`, + id, + attempt, + JSON.stringify({ questions: [{ question: `Choice ${id}-${attempt}?`, options: [] }] }), + attempt === 1 ? JSON.stringify({ text: `Answer ${id}` }) : null, + attempt === 1 ? user : null, + attempt, + attempt === 1 ? 10 : null, + attempt === 1 ? 11 : null, + ), + ); + } + await db.batch(statements); + return ids; +} + +describe("batched inbox reads", () => { + it("hydrates many jobs identically with a constant number of queries", async () => { + const observed = Array.from({ length: 120 }, (_, i) => `quest/job-${i}.md`); + const empty = counted(); + expect(await store.listInboxJobs(empty.db, 10, observed)).toEqual([]); + expect(empty.queries()).toBe(3); + const ids = await seed(120); + await project(11, user); + await seed(2, 11); + const expected = await Promise.all([...ids].reverse().map((id) => store.readJob(db, id))); + const all = counted(); + const result = await store.listInboxJobs(all.db, 10, observed); + expect(all.queries()).toBe(empty.queries()); + expect(result).toEqual(expected); + expect(result.find((job) => job.id === ids[1])).toMatchObject({ + attempt: 2, + result: { followups: ["Follow-up 1"] }, + events: [{ attempt: 2 }, { attempt: 1 }], + questions: [ + { attempt: 1, answer: { text: `Answer ${ids[1]}` }, answeredBy: user, answeredAt: 10, approvedAt: 11 }, + { attempt: 2, answer: null, answeredBy: null, answeredAt: null, approvedAt: null }, + ], + }); + expect(result.find((job) => job.id === ids[0])).toMatchObject({ events: [], questions: [] }); + const active = counted(); + expect(await store.listActivityJobs(active.db, 10, [])).toEqual( + expected.filter((job) => job!.state === "waiting"), + ); + expect(active.queries()).toBe(3); + const recent = counted(); + expect(await store.listJobs(recent.db, 10)).toEqual(expected.slice(0, 50)); + expect(recent.queries()).toBe(3); + }); + + it("keeps the entire inbox load's query count fixed as jobs grow", async () => { + const read = async () => { + const queries = counted(); + const result = await load({ ...env, DB: queries.db }, user); + return { queries: queries.queries(), result }; + }; + const empty = await read(); + await seed(120); + const many = await read(); + expect(many.queries).toBe(empty.queries); + expect(many.result.activity[0]).toHaveLength(60); + expect(many.result.items.filter((item) => item.job)).toHaveLength(60); + }); +}); diff --git a/quest/m1/README.md b/quest/m1/README.md index 9806a895..41c57488 100644 --- a/quest/m1/README.md +++ b/quest/m1/README.md @@ -19,6 +19,5 @@ additional login providers. This milestone does not gate the first release. - [Live inbox](/quest/m1/inbox-polish.md) - live agent cards, active quests on the log, runner choice for every job, expiring offers - [Worker dev setup](/quest/m1/worker-dev.md) - per-server cookies, seeded raw output, one fake GitHub, each test run alone - [Merge driver tuning](/quest/m1/resolve-tuning.md) - measure `quest resolve` on moq's history and match git's markers -- [Batched inbox reads](/quest/m1/inbox-reads.md) - a fixed number of D1 queries per project on each inbox load - [Incremental history](/quest/m1/history-pass.md) - each history pass reads only what changed since the last - [Gate owns descendants](/quest/m1/gate-owns-descendants.md) - cancelling `quest gate` leaves none of its command's processes behind, on Linux and macOS diff --git a/quest/m1/history-pass.md b/quest/m1/history-pass.md index 396e3c2c..980acbe1 100644 --- a/quest/m1/history-pass.md +++ b/quest/m1/history-pass.md @@ -16,7 +16,6 @@ Decided 2026-10-22 (user) as a follow-up of - Page by update time from the last pass's high-water mark, and read bodies only for open drafts, whose `## Decisions needed` section decides whether the board shows them as claimed or handed back. -- m1, next to [batched inbox reads](/quest/m1/inbox-reads.md): m0 pilots are - small enough that full passes are affordable. +- m1: m0 pilots are small enough that full passes are affordable. - Dropped (user): caching installation tokens across an inbox load, until pilots show slow inboxes. diff --git a/quest/m1/inbox-reads.md b/quest/m1/inbox-reads.md deleted file mode 100644 index e8f01b62..00000000 --- a/quest/m1/inbox-reads.md +++ /dev/null @@ -1,20 +0,0 @@ -# [S] Read the inbox's jobs in one query per project - -## Goal - -Loading the decision inbox runs a fixed number of D1 queries per project, -however many jobs, issues, PRs, and quests it shows. Today -`listInboxJobs` reads each returned job with its own `readJob` batch. - -## Plan - -Decided 2026-10-19 as a follow-up of the decision inbox (#144), whose final -review fix made the inbox return every observed target's latest job: - -- m1: m0 pilots track few targets per repository, and D1 calls within the - Worker are cheap, so this waits until after launch. -- Replace the per-job `readJob` batches with one joined query, or one batch - of a constant number of queries, per project. Keep the output identical, - including questions, answers, and attempt state. -- Add a test that seeds many jobs and asserts that the number of D1 queries - per inbox load does not grow with them.