Chunk D1 IN lists to stay under the 100-parameter cap

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<T extends SqlParam> {
+  /** `?, ?, …`: 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<T extends SqlParam>(values: readonly T[], otherParams = 0): InListChunk<T>[] {
+  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<T>[] = [];
+  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<T>(sql: string, ...params: SqlParam[]): Promise<T | null>;

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<PRRecord[]> {
-      if (repos.length === 0) return [];
-      const rows = await db.all<PRRow>(
-        `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<PRRow>(
+            `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<void> {
@@ -364,11 +366,16 @@ export function createPRStore(db: Db) {
 
     async deleteEvents(repo: string, number: number, ids: readonly number[]): Promise<void> {
       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 {

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");
+  });
+});

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<PRData> = {}) => 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", () => {

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([]);
+  });
+});

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"));

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<typeof prApp>;
 
+/** 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;