diff --git a/src/core/db.ts b/src/core/db.ts index 32d33c3..49c9642 100644 --- a/src/core/db.ts +++ b/src/core/db.ts @@ -1,5 +1,30 @@ export type SqlParam = string | number | null; +/** D1 binds at most this many parameters to one statement (stock SQLite allows thousands). */ +export const MAX_BOUND_PARAMS = 100; + +/** A slice of an `IN (…)` list that fits in one statement. */ +export interface InListChunk { + /** `?, ?, …`: one per value. */ + placeholders: string; + values: T[]; +} + +/** + * Splits the values of an `IN (…)` list into chunks that each fit in one statement alongside its + * `otherParams` other bound parameters. Run one statement per chunk; an empty list has no chunks. + */ +export function inListChunks(values: readonly T[], otherParams = 0): InListChunk[] { + const size = MAX_BOUND_PARAMS - otherParams; + if (size < 1) throw new RangeError(`No room for an IN list next to ${otherParams} other parameters`); + const chunks: InListChunk[] = []; + for (let start = 0; start < values.length; start += size) { + const chunk = values.slice(start, start + size); + chunks.push({ placeholders: chunk.map(() => "?").join(", "), values: chunk }); + } + return chunks; +} + /** Thin typed helpers over D1. Feature repositories write plain SQL on top of these. */ export interface Db { first(sql: string, ...params: SqlParam[]): Promise; diff --git a/src/features/pr_management/store.ts b/src/features/pr_management/store.ts index 88b7bfe..4c8bfa3 100644 --- a/src/features/pr_management/store.ts +++ b/src/features/pr_management/store.ts @@ -1,5 +1,5 @@ import { DateTime } from "luxon"; -import type { Db, SqlParam } from "../../core/db"; +import { type Db, inListChunks, type SqlParam } from "../../core/db"; import type { Mergeable } from "../../github"; import { decodeReviewers, type ReviewerLine } from "./reviewers"; import type { PRCategory, PRStatusState } from "./status"; @@ -321,13 +321,15 @@ export function createPRStore(db: Db) { /** Records not yet merged or closed in `repos`. */ async openRecords(repos: readonly string[]): Promise { - if (repos.length === 0) return []; - const rows = await db.all( - `SELECT * FROM pr_prs WHERE state NOT IN ${FINAL_STATES} AND repo IN (${repos.map(() => "?").join(", ")}) - ORDER BY repo, number`, - ...repos, + const rowsPerChunk = await Promise.all( + inListChunks(repos).map(({ placeholders, values }) => + db.all( + `SELECT * FROM pr_prs WHERE state NOT IN ${FINAL_STATES} AND repo IN (${placeholders})`, + ...values, + ), + ), ); - return rows.map(toRecord); + return rowsPerChunk.flat().sort(byRepoThenNumber).map(toRecord); }, async recordEvent(repo: string, number: number, event: NewPREvent): Promise { @@ -364,11 +366,16 @@ export function createPRStore(db: Db) { async deleteEvents(repo: string, number: number, ids: readonly number[]): Promise { if (ids.length === 0) return; - await db.run( - `DELETE FROM pr_events WHERE repo = ? AND number = ? AND id IN (${ids.map(() => "?").join(", ")})`, - repo, - number, - ...ids, + const prParams = [repo, number]; + // One transaction, so the events are consumed together or not at all. + await db.batch( + inListChunks(ids, prParams.length).map(({ placeholders, values }) => + db.statement( + `DELETE FROM pr_events WHERE repo = ? AND number = ? AND id IN (${placeholders})`, + ...prParams, + ...values, + ), + ), ); }, @@ -398,6 +405,12 @@ export function createPRStore(db: Db) { }; } +/** SQLite's `ORDER BY repo, number`: repos compare by code unit (they're ASCII), not by locale. */ +function byRepoThenNumber(a: PRRow, b: PRRow): number { + if (a.repo !== b.repo) return a.repo < b.repo ? -1 : 1; + return a.number - b.number; +} + function toRecord(row: PRRow): PRRecord { const at = (ms: number) => DateTime.fromMillis(ms, { zone: "utc" }); return { diff --git a/test/core/db.test.ts b/test/core/db.test.ts index 8e43d59..e0333f9 100644 --- a/test/core/db.test.ts +++ b/test/core/db.test.ts @@ -1,6 +1,6 @@ import { env } from "cloudflare:workers"; import { describe, expect, it } from "vitest"; -import { createDb } from "../../src/core/db"; +import { createDb, inListChunks, MAX_BOUND_PARAMS } from "../../src/core/db"; describe("Db", () => { const db = createDb(env.DB); @@ -45,3 +45,32 @@ describe("Db", () => { ).toEqual([{ changes: 1 }, { changes: 1 }]); }); }); + +describe("inListChunks", () => { + it("splits a list so each statement binds at most MAX_BOUND_PARAMS, other parameters included", () => { + const values = Array.from({ length: 250 }, (_, i) => i); + const chunks = inListChunks(values, 2); + expect(chunks.map((chunk) => chunk.values.length)).toEqual([98, 98, 54]); + expect(chunks.flatMap((chunk) => chunk.values)).toEqual(values); + expect(chunks.at(-1)?.placeholders).toBe(Array(54).fill("?").join(", ")); + }); + + it("keeps a short list in one chunk, and has none for an empty list", () => { + expect(inListChunks(["a", "b"])).toEqual([{ placeholders: "?, ?", values: ["a", "b"] }]); + expect(inListChunks([])).toEqual([]); + }); + + it("refuses when the other parameters leave no room", () => { + expect(() => inListChunks([1], MAX_BOUND_PARAMS)).toThrow(RangeError); + }); + + // The regression tests for long lists rely on the local D1 enforcing production's cap. + it("matches production D1, which binds MAX_BOUND_PARAMS parameters to a statement and no more", async () => { + const db = createDb(env.DB); + const count = (names: string[]) => + db.first(`SELECT COUNT(*) AS n FROM job_runs WHERE name IN (${names.map(() => "?").join(", ")})`, ...names); + const names = (n: number) => Array.from({ length: n }, (_, i) => `job-${i}`); + expect(await count(names(MAX_BOUND_PARAMS))).toEqual({ n: 0 }); + await expect(count(names(MAX_BOUND_PARAMS + 1))).rejects.toThrow("too many SQL variables"); + }); +}); diff --git a/test/features/pr_management/refresh.test.ts b/test/features/pr_management/refresh.test.ts index 8987ebb..70c0dfe 100644 --- a/test/features/pr_management/refresh.test.ts +++ b/test/features/pr_management/refresh.test.ts @@ -2,7 +2,7 @@ import { DateTime } from "luxon"; import { describe, expect, it, vi } from "vitest"; import type { PRData } from "../../../src/github"; import { aPR, aReview } from "../../builders/github"; -import { PR_CHANNEL, type PRTestApp, prApp, REPO } from "../../helpers/pr"; +import { MORE_THAN_D1_BINDS, PR_CHANNEL, type PRTestApp, prApp, REPO, recordCheckRuns } from "../../helpers/pr"; const awaitingBob = (overrides: Partial = {}) => aPR({ pendingReviewers: ["bob"], ...overrides }); @@ -164,6 +164,14 @@ describe("refresh: the record", () => { await h.refresh(101); expect(await h.events(101)).toEqual([]); }); + + it("consumes more events than D1 binds to one statement", async () => { + const h = prApp(); + h.github.upsert(aPR()); + await recordCheckRuns(h.store, MORE_THAN_D1_BINDS, h.clock.now()); + expect(await h.refresh(101)).toEqual({ kind: "ack" }); + expect(await h.events(101)).toEqual([]); + }); }); describe("refresh: data freshness", () => { diff --git a/test/features/pr_management/store.test.ts b/test/features/pr_management/store.test.ts new file mode 100644 index 0000000..f3d8615 --- /dev/null +++ b/test/features/pr_management/store.test.ts @@ -0,0 +1,52 @@ +import { env } from "cloudflare:workers"; +import { DateTime } from "luxon"; +import { describe, expect, it } from "vitest"; +import { createDb } from "../../../src/core/db"; +import { createPRStore } from "../../../src/features/pr_management/store"; +import { aRecord } from "../../builders/pr_record"; +import { MORE_THAN_D1_BINDS, REPO, recordCheckRuns } from "../../helpers/pr"; + +const store = () => createPRStore(createDb(env.DB)); +const NOW = DateTime.fromISO("2026-10-05T14:00:00Z", { zone: "utc" }); + +describe("PR store: lists longer than D1 binds to one statement", () => { + it("deletes exactly the given events of the PR", async () => { + const s = store(); + await recordCheckRuns(s, MORE_THAN_D1_BINDS + 1, NOW); + await recordCheckRuns(s, 1, NOW, 102); + const [kept, ...consumed] = await s.events(REPO, 101, NOW); + + await s.deleteEvents( + REPO, + 101, + consumed.map((event) => event.id), + ); + + expect(await s.events(REPO, 101, NOW)).toEqual([kept]); + expect(await s.events(REPO, 102, NOW)).toHaveLength(1); + }); + + it("reads the open records of every repo, ordered by repo then number", async () => { + const s = store(); + const repos = Array.from({ length: MORE_THAN_D1_BINDS }, (_, i) => `Kiln-AI/repo-${String(i).padStart(3, "0")}`); + // Inserted last repo first, so the order can only come from sorting across the chunks. + for (const repo of repos.toReversed()) { + await s.insert(aRecord({ repo, number: 2 })); + await s.insert(aRecord({ repo, number: 1 })); + } + await s.insert(aRecord({ repo: repos[0], number: 3, state: "merged" })); + await s.insert(aRecord({ repo: "Kiln-AI/untracked", number: 1 })); + + const records = await s.openRecords(repos.toReversed()); + + expect(records.map(({ repo, number }) => `${repo}#${number}`)).toEqual( + repos.flatMap((repo) => [`${repo}#1`, `${repo}#2`]), + ); + }); + + it("reads no records for no repos", async () => { + const s = store(); + await s.insert(aRecord()); + expect(await s.openRecords([])).toEqual([]); + }); +}); diff --git a/test/features/pr_management/sweep.test.ts b/test/features/pr_management/sweep.test.ts index 3add975..3d6a8e8 100644 --- a/test/features/pr_management/sweep.test.ts +++ b/test/features/pr_management/sweep.test.ts @@ -2,7 +2,7 @@ import type { DateTime } from "luxon"; import { describe, expect, it, vi } from "vitest"; import { RateLimitedError } from "../../../src/github"; import { aPR, aReview } from "../../builders/github"; -import { PR_CHANNEL, type PRTestApp, prApp, REPO } from "../../helpers/pr"; +import { MORE_THAN_D1_BINDS, PR_CHANNEL, type PRTestApp, prApp, REPO, recordCheckRuns } from "../../helpers/pr"; const TOP_OF_HOUR = Date.parse("2026-10-05T15:00:00Z"); const SWEEP_SOURCE = "pr_management.sweep"; @@ -137,6 +137,18 @@ describe("hourly sweep", () => { expect(h.log.at("warn").map((e) => e.msg)).toContain("GitHub rate limit hit; sweep stopped early"); }); + it("consumes more events on a PR than D1 binds to one statement", async () => { + const h = prApp(); + h.github.upsert(aPR({ pendingReviewers: ["bob"] })); + await recordCheckRuns(h.store, MORE_THAN_D1_BINDS, h.clock.now()); + + await runSweep(h); + + expect(reported(h)).toEqual([]); + expect(await h.events(101)).toEqual([]); + expect(await h.record(101)).toMatchObject({ state: "awaiting_review" }); + }); + it("reports a failed sweep query through the scheduler", async () => { const h = prApp(); h.github.fail(new Error("github is down")); diff --git a/test/helpers/pr.ts b/test/helpers/pr.ts index cfdd037..58115b8 100644 --- a/test/helpers/pr.ts +++ b/test/helpers/pr.ts @@ -1,11 +1,13 @@ +import type { DateTime } from "luxon"; import type { AppOverrides } from "../../src/core/app"; import { loadConfig } from "../../src/core/config"; +import { MAX_BOUND_PARAMS } from "../../src/core/db"; import { normalizeGithubLogin, type UserDirectory } from "../../src/core/directory"; import { prManagement } from "../../src/features/pr_management"; import type { PRConfig } from "../../src/features/pr_management/config"; import { createPeople, type People } from "../../src/features/pr_management/people"; import type { PRContext } from "../../src/features/pr_management/refresh"; -import { createPRStore } from "../../src/features/pr_management/store"; +import { createPRStore, type PRStore } from "../../src/features/pr_management/store"; import { aConfig } from "../builders/config"; import { fakeBatch, fakeMessage } from "../fakes/batch"; import { testApp } from "./app"; @@ -61,6 +63,22 @@ export function prApp(overrides: AppOverrides = {}) { export type PRTestApp = ReturnType; +/** More values than D1 binds to one statement, or to two. */ +export const MORE_THAN_D1_BINDS = MAX_BOUND_PARAMS * 2 + 50; + +/** Records `count` check_run events on a PR, as a push to a big CI matrix does. */ +export async function recordCheckRuns(store: PRStore, count: number, receivedAt: DateTime, number = 101, repo = REPO) { + for (let i = 0; i < count; i++) { + await store.recordEvent(repo, number, { + event: "check_run", + action: "completed", + actor: "github-actions[bot]", + subject: null, + receivedAt, + }); + } +} + /** The feature's context over the test app, for calling its functions directly. */ export function prContext(h: PRTestApp): PRContext { const config = loadConfig(prConfig(), "development", [prManagement]).features.get("pr_management") as PRConfig;