perf(db): instance-wide completed-batch sweep commits in 200-batch chunks (SEC-D)

Refs #12969

src/lib/db/batches.ts

@@ -443,10 +443,23 @@ export type DeleteCompletedBatchesScope = { apiKeyId: string } | { allTenants: t
  * the caller's are soft-deleted; a referenced file another tenant owns (or an
  * unowned one) is left intact and is not counted in deletedFiles.
  *
- * The file soft-deletes, the checkpoint DELETE and the batches DELETE run in one
- * transaction, so a mid-sweep failure rolls everything back — no batch row is
- * left pointing at a file whose content was already nulled.
+ * The file soft-deletes, the checkpoint DELETE and the batches DELETE for a set
+ * of batch ids run in one transaction, so a mid-sweep failure rolls that set back
+ * — no batch row is left pointing at a file whose content was already nulled.
+ * Key mode runs that unit once over every completed batch the key owns.
+ * Instance mode (`allTenants`) runs it per chunk of `INSTANCE_SWEEP_CHUNK` ids
+ * (SEC-D): a large sweep never holds one write lock over the whole table, each
+ * chunk stays atomic, and a failure inside chunk N leaves chunks < N committed,
+ * chunk N fully rolled back, and rethrows. The returned totals sum the chunks.
+ *
+ * The ids of a unit are bound as `IN (?, …)` placeholders. A chunk is far below
+ * SQLite's default SQLITE_MAX_VARIABLE_NUMBER (32766 since 3.32); the key-mode
+ * list is bounded by that key's completed batches — should a single key ever
+ * own more than ~32k completed batches, chunk key mode the same way.
  */
+/** Instance-wide sweeps commit in chunks of this many batches (SEC-D). */
+export const INSTANCE_SWEEP_CHUNK = 200;
+
 export function deleteCompletedBatches(scope: DeleteCompletedBatchesScope): {
   deletedBatches: number;
   deletedFiles: number;
@@ -463,16 +476,18 @@ export function deleteCompletedBatches(scope: DeleteCompletedBatchesScope): {
 
   const db = getDbInstance();
 
-  const ownershipClause = allTenants ? "" : " AND api_key_id = ?";
-  const ownershipArgs = allTenants ? [] : [apiKeyId];
-
-  const sweep = db.transaction(() => {
-    // Collect unique file IDs from the completed batches in scope
+  // One consistent unit: file soft-deletes → checkpoints → batch rows for a
+  // given set of batch ids. Key mode runs it once over every completed batch
+  // the key owns; instance mode runs it per chunk so a large sweep never holds
+  // one write-lock for the whole table (SEC-D) while each chunk stays atomic.
+  const sweepIds = db.transaction((ids: string[]) => {
+    if (ids.length === 0) return { deletedBatches: 0, deletedFiles: 0 };
+    const marks = ids.map(() => "?").join(",");
     const rows = db
       .prepare(
-        `SELECT input_file_id, output_file_id, error_file_id FROM batches WHERE status = 'completed'${ownershipClause}`
+        `SELECT input_file_id, output_file_id, error_file_id FROM batches WHERE id IN (${marks})`
       )
-      .all(...ownershipArgs) as Array<{
+      .all(...ids) as Array<{
       input_file_id: string | null;
       output_file_id: string | null;
       error_file_id: string | null;
@@ -501,15 +516,32 @@ export function deleteCompletedBatches(scope: DeleteCompletedBatchesScope): {
       }
     }
 
-    db.prepare(
-      `DELETE FROM batch_item_checkpoints WHERE batch_id IN (SELECT id FROM batches WHERE status = 'completed'${ownershipClause})`
-    ).run(...ownershipArgs);
-
-    const result = db
-      .prepare(`DELETE FROM batches WHERE status = 'completed'${ownershipClause}`)
-      .run(...ownershipArgs);
+    db.prepare(`DELETE FROM batch_item_checkpoints WHERE batch_id IN (${marks})`).run(...ids);
+    const result = db.prepare(`DELETE FROM batches WHERE id IN (${marks})`).run(...ids);
     return { deletedBatches: result.changes, deletedFiles };
   });
 
-  return sweep();
+  if (!allTenants) {
+    const ids = (
+      db
+        .prepare(
+          "SELECT id FROM batches WHERE status = 'completed' AND api_key_id = ? ORDER BY rowid"
+        )
+        .all(apiKeyId) as Array<{ id: string }>
+    ).map((r) => r.id);
+    return sweepIds(ids);
+  }
+
+  const totals = { deletedBatches: 0, deletedFiles: 0 };
+  const nextChunk = db.prepare(
+    "SELECT id FROM batches WHERE status = 'completed' ORDER BY rowid LIMIT ?"
+  );
+  for (;;) {
+    const ids = (nextChunk.all(INSTANCE_SWEEP_CHUNK) as Array<{ id: string }>).map((r) => r.id);
+    if (ids.length === 0) break;
+    const part = sweepIds(ids);
+    totals.deletedBatches += part.deletedBatches;
+    totals.deletedFiles += part.deletedFiles;
+  }
+  return totals;
 }

tests/unit/batches-delete-completed-ownership-wvxc.test.ts

@@ -37,7 +37,7 @@ process.env.DATA_DIR = TEST_DATA_DIR;
 process.env.APP_LOG_LEVEL = "warn";
 
 const { createFile, getFile, getFileContent } = await import("../../src/lib/db/files.ts");
-const { createBatch, getBatch, deleteCompletedBatches } =
+const { createBatch, getBatch, deleteCompletedBatches, INSTANCE_SWEEP_CHUNK } =
   await import("../../src/lib/db/batches.ts");
 const { getDbInstance, resetDbInstance } = await import("../../src/lib/db/core.ts");
 
@@ -296,4 +296,60 @@ describe("deleteCompletedBatches — ownership boundary (GHSA-wvxc-jp3v-5mg5)",
     assert.ok(logged, "the failure is logged at warn level");
     assert.ok(logged!.includes(own.file.id), "the log line names the file id");
   });
+  it("SEC-D: the instance sweep runs in chunks of INSTANCE_SWEEP_CHUNK and still deletes everything", () => {
+    const total = INSTANCE_SWEEP_CHUNK * 2 + 50; // 3 chunks: 200 + 200 + 50
+    const ids: string[] = [];
+    for (let i = 0; i < total; i++)
+      ids.push(seedCompletedBatch(i % 2 ? "key-chunk-a" : null, `chunk-${i}`).batch.id);
+    // `db.transaction(fn)` is called ONCE to build the unit; what must happen per
+    // chunk is the INVOCATION of the unit — count those.
+    const db = getDbInstance();
+    let runs = 0;
+    const origTx = db.transaction.bind(db);
+    const txSpy = mock.method(db, "transaction", (fn: (...a: unknown[]) => unknown) => {
+      const tx = origTx(fn);
+      return (...args: unknown[]) => {
+        runs++;
+        return tx(...args);
+      };
+    });
+
+    let result: ReturnType<typeof deleteCompletedBatches>;
+    try {
+      result = deleteCompletedBatches({ allTenants: true });
+    } finally {
+      txSpy.mock.restore();
+    }
+
+    assert.strictEqual(result.deletedBatches, total);
+    assert.strictEqual(result.deletedFiles, total);
+    assert.strictEqual(runs, 3, "one transaction per chunk (200 + 200 + 50)");
+    for (const id of ids) assert.strictEqual(getBatch(id), null);
+  });
+
+  it("SEC-D: a failure in chunk 2 keeps chunk 1 done and rolls chunk 2 back entirely", () => {
+    const first = Array.from({ length: INSTANCE_SWEEP_CHUNK }, (_, i) =>
+      seedCompletedBatch(null, `c1-${i}`)
+    );
+    const second = Array.from({ length: 10 }, (_, i) => seedCompletedBatch(null, `c2-${i}`));
+    const poison = second[5].batch.id;
+    const db = getDbInstance();
+    db.exec(
+      `CREATE TRIGGER wvxc_chunk_poison BEFORE DELETE ON batches WHEN OLD.id = '${poison}' BEGIN SELECT RAISE(ABORT, 'poison'); END`
+    );
+    try {
+      assert.throws(() => deleteCompletedBatches({ allTenants: true }), /poison/);
+    } finally {
+      db.exec("DROP TRIGGER IF EXISTS wvxc_chunk_poison");
+    }
+    for (const s of first) assert.strictEqual(getBatch(s.batch.id), null, "chunk 1 committed");
+    for (const s of second) {
+      assert.ok(getBatch(s.batch.id), "chunk 2 rolled back as a unit");
+      assert.strictEqual(
+        getFileContent(s.file.id)?.toString(),
+        s.file.filename.replace(".jsonl", ""),
+        "chunk 2 file content restored"
+      );
+    }
+  });
 });