diff --git a/open-sse/services/batchProcessor.ts b/open-sse/services/batchProcessor.ts index 9cde8a0321..40818df4d4 100644 --- a/open-sse/services/batchProcessor.ts +++ b/open-sse/services/batchProcessor.ts @@ -1,14 +1,20 @@ import { v4 as uuidv4 } from "uuid"; -import type { BatchRecord } from "@/lib/localDb"; +import type { BatchItemCheckpoint, BatchRecord } from "@/lib/localDb"; import { + countBatchItemCheckpoints, createFile, deleteFile, + ensureBatchItemCheckpoints, getApiKeyById, getBatch, getFileContent, getPendingBatches, getTerminalBatches, + listBatchItemCheckpoints, listFiles, + markBatchItemError, + markBatchItemProcessing, + markBatchItemResult, updateBatch, } from "@/lib/localDb"; import { dispatch } from "@/lib/batches/dispatch"; @@ -67,28 +73,12 @@ export async function processPendingBatches(): Promise { const pending = getPendingBatches(); // Phase 1: Stale recovery — in_progress/finalizing batches not in activeBatches - // are from a previous session; reset them to validating so they get picked up fresh + // are from a previous session; reset checkpointed batches to validating so they + // can be completed without replaying already-dispatched items. for (const batch of pending) { if (batch.status === "in_progress" || batch.status === "finalizing") { if (!activeBatches.has(batch.id)) { - console.log(`[BATCH] Recovering stale batch ${batch.id} (${batch.status}) → validating`); - - if (batch.outputFileId) { - deleteFile(batch.outputFileId); - } - if (batch.errorFileId) { - deleteFile(batch.errorFileId); - } - - updateBatch(batch.id, { - status: "validating", - inProgressAt: null, - finalizingAt: null, - outputFileId: null, - errorFileId: null, - requestCountsCompleted: 0, - requestCountsFailed: 0, - }); + recoverStaleBatch(batch); } } } @@ -116,6 +106,49 @@ export async function processPendingBatches(): Promise { await cleanupExpiredBatches(); } +function recoverStaleBatch(batch: BatchRecord): void { + const checkpointCount = countBatchItemCheckpoints(batch.id); + const hasPotentialExternalEffects = + batch.requestCountsTotal > 0 || + batch.requestCountsCompleted > 0 || + batch.requestCountsFailed > 0 || + batch.status === "finalizing"; + + if (checkpointCount === 0 && hasPotentialExternalEffects) { + console.warn( + `[BATCH] Stale batch ${batch.id} has no item checkpoints; failing instead of replaying provider calls` + ); + failBatch( + batch.id, + "Cannot safely recover stale batch because item checkpoints are unavailable; create a new batch to retry intentionally." + ); + return; + } + + console.log(`[BATCH] Recovering stale batch ${batch.id} (${batch.status}) → validating`); + + if (batch.outputFileId) { + deleteFile(batch.outputFileId); + } + if (batch.errorFileId) { + deleteFile(batch.errorFileId); + } + + updateBatch(batch.id, { + status: "validating", + inProgressAt: null, + finalizingAt: null, + outputFileId: null, + errorFileId: null, + ...(checkpointCount === 0 + ? { + requestCountsCompleted: 0, + requestCountsFailed: 0, + } + : {}), + }); +} + function parseBatchWindowSeconds(window: string | null | undefined): number { if (!window) return DEFAULT_BATCH_WINDOW_SECONDS; const match = /^(\d+)([hdm])$/.exec(window); @@ -285,6 +318,8 @@ async function startBatch(batch: any): Promise { console.log(`[BATCH] Batch ${batch.id} contains (${total} items)`); + ensureBatchItemCheckpoints(batch.id, parsedItems.items); + updateBatch(batch.id, { status: "in_progress", inProgressAt: Math.floor(Date.now() / 1000), @@ -316,12 +351,21 @@ const HEADERS_CACHE_TTL_MS = 60_000; async function processBatchItems(batch: BatchRecord, items: BatchRequestItem[]): Promise { const state = createBatchState(batch); + const checkpoints = new Map( + listBatchItemCheckpoints(batch.id).map((checkpoint) => [checkpoint.lineNumber, checkpoint]) + ); const apiKey = await resolveApiKey(batch); for (const item of items) { if (isBatchCancelled(batch.id)) break; + const checkpoint = checkpoints.get(item.lineNumber); + if (checkpoint && applyRecoveredCheckpoint(batch.id, item, checkpoint, state)) { + maybePersistProgress(batch.id, state); + continue; + } + const cachedHeaders = prevHeaders && Date.now() - prevHeadersTimestamp < HEADERS_CACHE_TTL_MS ? prevHeaders : null; if (cachedHeaders) { @@ -331,6 +375,8 @@ async function processBatchItems(batch: BatchRecord, items: BatchRequestItem[]): } } + markBatchItemProcessing(batch.id, item); + try { const response = await processSingleItemWithRetry(item, apiKey); let responseBody: unknown; @@ -352,13 +398,17 @@ async function processBatchItems(batch: BatchRecord, items: BatchRequestItem[]): }, }; + markBatchItemResult(batch.id, item, wrapped); state.results.push(wrapped); applyItemResult(state, response.status, responseBody); prevHeaders = response.headers; prevHeadersTimestamp = Date.now(); } catch (exception) { // Track processing-level errors separately (items that failed to be processed) - state.errors.push({ custom_id: item.customId ?? null, error: String(exception) }); + const error = { custom_id: item.customId ?? null, error: String(exception) }; + markBatchItemError(batch.id, item, error); + state.errors.push(error); + state.failed++; prevHeaders = null; prevHeadersTimestamp = 0; } @@ -369,6 +419,43 @@ async function processBatchItems(batch: BatchRecord, items: BatchRequestItem[]): return finalizeBatch(batch.id, state.results, state.errors); } +function applyRecoveredCheckpoint( + batchId: string, + item: BatchRequestItem, + checkpoint: BatchItemCheckpoint, + state: ReturnType +): boolean { + if (checkpoint.status === "completed" && checkpoint.result) { + state.results.push(checkpoint.result); + applyItemResult( + state, + checkpoint.result.response?.status_code ?? 500, + checkpoint.result.response?.body + ); + return true; + } + + if (checkpoint.status === "errored" && checkpoint.error) { + state.errors.push(checkpoint.error); + state.failed++; + return true; + } + + if (checkpoint.status === "processing") { + const error = { + custom_id: item.customId ?? null, + error: + "Batch item was interrupted before its provider response was recorded; it was not replayed to avoid duplicate provider work.", + }; + markBatchItemError(batchId, item, error); + state.errors.push(error); + state.failed++; + return true; + } + + return false; +} + function isBatchCancelled(batchId: string): boolean { const current = getBatch(batchId); diff --git a/src/lib/db/batches.ts b/src/lib/db/batches.ts index 336940c345..268f8aa4a9 100644 --- a/src/lib/db/batches.ts +++ b/src/lib/db/batches.ts @@ -82,6 +82,42 @@ export interface BatchRecord { outputExpiresAfterAnchor?: string | null; } +export type BatchItemCheckpointStatus = "pending" | "processing" | "completed" | "errored"; + +export interface BatchItemCheckpoint { + batchId: string; + lineNumber: number; + customId: string | null; + status: BatchItemCheckpointStatus; + result: any | null; + error: any | null; + createdAt: number; + updatedAt: number; +} + +function parseJsonColumn(value: unknown): any | null { + if (value == null) return null; + if (typeof value !== "string") return value; + try { + return JSON.parse(value); + } catch { + return null; + } +} + +function parseBatchItemCheckpoint(row: any): BatchItemCheckpoint { + return { + batchId: row.batch_id, + lineNumber: Number(row.line_number), + customId: row.custom_id ?? null, + status: row.status, + result: parseJsonColumn(row.result_json), + error: parseJsonColumn(row.error_json), + createdAt: Number(row.created_at), + updatedAt: Number(row.updated_at), + }; +} + export function createBatch( batch: Omit< BatchRecord, @@ -156,6 +192,128 @@ export function updateBatch(id: string, updates: Partial): boolean return result.changes > 0; } +export function ensureBatchItemCheckpoints( + batchId: string, + items: Array<{ lineNumber: number; customId: string | null }> +): void { + if (items.length === 0) return; + + const db = getDbInstance(); + const now = Math.floor(Date.now() / 1000); + const insert = db.prepare(` + INSERT OR IGNORE INTO batch_item_checkpoints ( + batch_id, + line_number, + custom_id, + status, + result_json, + error_json, + created_at, + updated_at + ) + VALUES (?, ?, ?, 'pending', NULL, NULL, ?, ?) + `); + + const tx = db.transaction(() => { + for (const item of items) { + insert.run(batchId, item.lineNumber, item.customId, now, now); + } + }); + tx(); +} + +export function countBatchItemCheckpoints(batchId: string): number { + const db = getDbInstance(); + const row = db + .prepare("SELECT COUNT(*) AS c FROM batch_item_checkpoints WHERE batch_id = ?") + .get(batchId) as { c: number } | undefined; + return row ? Number(row.c) : 0; +} + +export function listBatchItemCheckpoints(batchId: string): BatchItemCheckpoint[] { + const db = getDbInstance(); + const rows = db + .prepare( + ` + SELECT batch_id, line_number, custom_id, status, result_json, error_json, created_at, updated_at + FROM batch_item_checkpoints + WHERE batch_id = ? + ORDER BY line_number ASC + ` + ) + .all(batchId); + return rows.map((row) => parseBatchItemCheckpoint(row)); +} + +export function markBatchItemProcessing( + batchId: string, + item: { lineNumber: number; customId: string | null } +): void { + const db = getDbInstance(); + const now = Math.floor(Date.now() / 1000); + db.prepare( + ` + INSERT INTO batch_item_checkpoints ( + batch_id, + line_number, + custom_id, + status, + result_json, + error_json, + created_at, + updated_at + ) + VALUES (?, ?, ?, 'processing', NULL, NULL, ?, ?) + ON CONFLICT(batch_id, line_number) DO UPDATE SET + custom_id = excluded.custom_id, + status = 'processing', + result_json = NULL, + error_json = NULL, + updated_at = excluded.updated_at + ` + ).run(batchId, item.lineNumber, item.customId, now, now); +} + +export function markBatchItemResult( + batchId: string, + item: { lineNumber: number; customId: string | null }, + result: any +): void { + const db = getDbInstance(); + const now = Math.floor(Date.now() / 1000); + db.prepare( + ` + UPDATE batch_item_checkpoints + SET custom_id = ?, + status = 'completed', + result_json = ?, + error_json = NULL, + updated_at = ? + WHERE batch_id = ? AND line_number = ? + ` + ).run(item.customId, JSON.stringify(result), now, batchId, item.lineNumber); +} + +export function markBatchItemError( + batchId: string, + item: { lineNumber: number; customId: string | null }, + error: any +): void { + const db = getDbInstance(); + const now = Math.floor(Date.now() / 1000); + db.prepare( + ` + UPDATE batch_item_checkpoints + SET custom_id = ?, + status = 'errored', + result_json = NULL, + error_json = ?, + updated_at = ? + WHERE batch_id = ? AND line_number = ? + ` + ).run(item.customId, JSON.stringify(error), now, batchId, item.lineNumber); +} + export function listBatches(apiKeyId?: string, limit: number = 20, after?: string): BatchRecord[] { const db = getDbInstance(); const afterBatch = after ? getBatch(after) : null; @@ -224,6 +382,8 @@ export function deleteBatch(id: string): boolean { const batch = getBatch(id); if (!batch) return false; + db.prepare("DELETE FROM batch_item_checkpoints WHERE batch_id = ?").run(id); + // Soft-delete associated files (input, output, error) if (batch.inputFileId) { try { @@ -281,6 +441,10 @@ export function deleteCompletedBatches(): { deletedBatches: number; deletedFiles } } + db.prepare( + "DELETE FROM batch_item_checkpoints WHERE batch_id IN (SELECT id FROM batches WHERE status = 'completed')" + ).run(); + const result = db.prepare("DELETE FROM batches WHERE status = 'completed'").run(); return { deletedBatches: result.changes, deletedFiles }; } diff --git a/src/lib/db/migrations/112_batch_item_checkpoints.sql b/src/lib/db/migrations/112_batch_item_checkpoints.sql new file mode 100644 index 0000000000..e93917f242 --- /dev/null +++ b/src/lib/db/migrations/112_batch_item_checkpoints.sql @@ -0,0 +1,18 @@ +-- 110_batch_item_checkpoints.sql +-- Durable per-item checkpoints for OpenAI-compatible batch processing. + +CREATE TABLE IF NOT EXISTS batch_item_checkpoints ( + batch_id TEXT NOT NULL, + line_number INTEGER NOT NULL, + custom_id TEXT, + status TEXT NOT NULL CHECK(status IN ('pending', 'processing', 'completed', 'errored')), + result_json TEXT, + error_json TEXT, + created_at INTEGER NOT NULL, + updated_at INTEGER NOT NULL, + PRIMARY KEY (batch_id, line_number), + FOREIGN KEY(batch_id) REFERENCES batches(id) ON DELETE CASCADE +); + +CREATE INDEX IF NOT EXISTS idx_batch_item_checkpoints_batch_status + ON batch_item_checkpoints(batch_id, status); diff --git a/src/lib/localDb.ts b/src/lib/localDb.ts index 98a24aecb7..6390ff3808 100755 --- a/src/lib/localDb.ts +++ b/src/lib/localDb.ts @@ -290,12 +290,18 @@ export { countBatches, getPendingBatches, getTerminalBatches, + ensureBatchItemCheckpoints, + countBatchItemCheckpoints, + listBatchItemCheckpoints, + markBatchItemProcessing, + markBatchItemResult, + markBatchItemError, deleteBatch, deleteCompletedBatches, } from "./db/batches"; export type { FileRecord } from "./db/files"; -export type { BatchRecord } from "./db/batches"; +export type { BatchItemCheckpoint, BatchRecord } from "./db/batches"; export type { ModelComboMapping } from "./db/modelComboMappings"; @@ -578,12 +584,7 @@ export { listAllocationsForApiKey, } from "./db/quotaPools"; // Quota per-(key, model) caps — Group B Fase 3 #7 -export { - getModelCap, - listModelCaps, - setModelCap, - deleteModelCap, -} from "./db/quotaModelCaps"; +export { getModelCap, listModelCaps, setModelCap, deleteModelCap } from "./db/quotaModelCaps"; export { // Quota Groups (B2) diff --git a/tests/unit/batch-processor.test.ts b/tests/unit/batch-processor.test.ts index 18903c6ac5..b96df59ffb 100644 --- a/tests/unit/batch-processor.test.ts +++ b/tests/unit/batch-processor.test.ts @@ -319,10 +319,147 @@ test("processPendingBatches caches rate-limit headers across sequential batches" assert.strictEqual(afterReset.timestamp, 0, "reset should clear cached timestamp"); }); -test("processPendingBatches should recover stale in_progress batches", async () => { +test("processPendingBatches should recover checkpointed stale batches without replaying completed items", async () => { + const lines = [ + JSON.stringify({ + custom_id: "already-done", + method: "POST", + url: "/v1/chat/completions", + body: { model: "gpt-4", messages: [{ role: "user", content: "first" }] }, + }), + JSON.stringify({ + custom_id: "needs-dispatch", + method: "POST", + url: "/v1/chat/completions", + body: { model: "gpt-4", messages: [{ role: "user", content: "second" }] }, + }), + ]; + const file = await localDb.createFile({ + bytes: Buffer.byteLength(lines.join("\n") + "\n"), + filename: "checkpointed_stale_test.jsonl", + purpose: "batch_input", + content: Buffer.from(lines.join("\n") + "\n"), + }); + + const batch = await localDb.createBatch({ + endpoint: "/v1/chat/completions", + status: "in_progress", + inputFileId: file.id, + completionWindow: "24h", + inProgressAt: Math.floor(Date.now() / 1000), + }); + await localDb.updateBatch(batch.id, { + requestCountsTotal: 2, + requestCountsCompleted: 1, + }); + await localDb.ensureBatchItemCheckpoints(batch.id, [ + { lineNumber: 1, customId: "already-done" }, + { lineNumber: 2, customId: "needs-dispatch" }, + ]); + await localDb.markBatchItemResult( + batch.id, + { lineNumber: 1, customId: "already-done" }, + { + id: "req_checkpointed", + custom_id: "already-done", + response: { + status_code: 200, + body: { + id: "chatcmpl-checkpointed", + choices: [{ message: { content: "from checkpoint" } }], + usage: { prompt_tokens: 10, completion_tokens: 5 }, + }, + }, + } + ); + + let callCount = 0; + mock.method(dispatch, "dispatchBatchApiRequest", async () => { + callCount++; + return new Response( + JSON.stringify({ + id: "chatcmpl-live", + choices: [{ message: { content: "processed once" } }], + usage: { prompt_tokens: 8, completion_tokens: 4 }, + }), + { status: 200, headers: { "Content-Type": "application/json" } } + ); + }); + + await batchProcessor.processPendingBatches(); + + await waitForAllBatches(); + + const updated = await localDb.getBatch(batch.id); + assert.strictEqual(updated?.status, "completed", "Stale batch should be recovered and completed"); + assert.strictEqual(callCount, 1, "only the unchecked item should be dispatched"); + assert.strictEqual(updated?.requestCountsCompleted, 2); + assert.ok(updated?.outputFileId, "recovered batch should emit an output file"); + + const output = localDb.getFileContent(updated!.outputFileId!); + assert.ok(output, "output file content should exist"); + const outputRows = output + .toString() + .trim() + .split("\n") + .map((line) => JSON.parse(line)); + assert.deepStrictEqual( + outputRows.map((row) => row.custom_id), + ["already-done", "needs-dispatch"] + ); +}); + +test("processPendingBatches should not replay interrupted checkpoint items", async () => { const file = await localDb.createFile({ bytes: 10, - filename: "stale_test.jsonl", + filename: "stale_processing_checkpoint.jsonl", + purpose: "batch_input", + content: Buffer.from( + JSON.stringify({ + custom_id: "in-flight", + method: "POST", + url: "/v1/chat/completions", + body: { model: "gpt-4", messages: [{ role: "user", content: "hi" }] }, + }) + "\n" + ), + }); + + const batch = await localDb.createBatch({ + endpoint: "/v1/chat/completions", + status: "in_progress", + inputFileId: file.id, + completionWindow: "24h", + inProgressAt: Math.floor(Date.now() / 1000), + }); + await localDb.updateBatch(batch.id, { requestCountsTotal: 1 }); + await localDb.ensureBatchItemCheckpoints(batch.id, [{ lineNumber: 1, customId: "in-flight" }]); + await localDb.markBatchItemProcessing(batch.id, { lineNumber: 1, customId: "in-flight" }); + + let callCount = 0; + mock.method(dispatch, "dispatchBatchApiRequest", async () => { + callCount++; + throw new Error("interrupted checkpoint should not dispatch"); + }); + + await batchProcessor.processPendingBatches(); + + await waitForAllBatches(); + + const updated = await localDb.getBatch(batch.id); + assert.strictEqual(updated?.status, "completed"); + assert.strictEqual(callCount, 0, "interrupted checkpoint should not be replayed"); + assert.strictEqual(updated?.requestCountsFailed, 1); + assert.ok(updated?.errorFileId, "interrupted item should be emitted as an error row"); + + const errorOutput = localDb.getFileContent(updated!.errorFileId!); + assert.ok(errorOutput, "error file content should exist"); + assert.match(errorOutput.toString(), /not replayed to avoid duplicate provider work/); +}); + +test("processPendingBatches should fail stale batches without checkpoints instead of replaying", async () => { + const file = await localDb.createFile({ + bytes: 10, + filename: "legacy_stale_no_checkpoints.jsonl", purpose: "batch_input", content: Buffer.from( JSON.stringify({ @@ -340,77 +477,20 @@ test("processPendingBatches should recover stale in_progress batches", async () completionWindow: "24h", inProgressAt: Math.floor(Date.now() / 1000), }); + await localDb.updateBatch(batch.id, { requestCountsTotal: 1 }); + let callCount = 0; mock.method(dispatch, "dispatchBatchApiRequest", async () => { - return new Response( - JSON.stringify({ - id: "chatcmpl-stale", - choices: [{ message: { content: "recovered" } }], - usage: { prompt_tokens: 10, completion_tokens: 5 }, - }), - { status: 200, headers: { "Content-Type": "application/json" } } - ); + callCount++; + throw new Error("legacy stale batch should not dispatch"); }); await batchProcessor.processPendingBatches(); - await waitForAllBatches(); - const updated = await localDb.getBatch(batch.id); - assert.strictEqual(updated?.status, "completed", "Stale batch should be recovered and completed"); - assert.ok( - updated?.inProgressAt != null, - "inProgressAt should be set (from fresh start, not stale)" - ); -}); - -test("processPendingBatches should recover stale finalizing batches", async () => { - const file = await localDb.createFile({ - bytes: 10, - filename: "stale_finalizing.jsonl", - purpose: "batch_input", - content: Buffer.from( - JSON.stringify({ - method: "POST", - url: "/v1/chat/completions", - body: { model: "gpt-4", messages: [{ role: "user", content: "hi" }] }, - }) + "\n" - ), - }); - - const batch = await localDb.createBatch({ - endpoint: "/v1/chat/completions", - status: "finalizing", - inputFileId: file.id, - completionWindow: "24h", - finalizingAt: Math.floor(Date.now() / 1000), - }); - - mock.method(dispatch, "dispatchBatchApiRequest", async () => { - return new Response( - JSON.stringify({ - id: "chatcmpl-stale-finalizing", - choices: [{ message: { content: "recovered" } }], - usage: { prompt_tokens: 10, completion_tokens: 5 }, - }), - { status: 200, headers: { "Content-Type": "application/json" } } - ); - }); - - await batchProcessor.processPendingBatches(); - - await waitForAllBatches(); - - const updated = await localDb.getBatch(batch.id); - assert.strictEqual( - updated?.status, - "completed", - "Stale finalizing batch should be recovered and completed" - ); - assert.ok( - updated?.inProgressAt != null, - "inProgressAt should be set by fresh start after recovery" - ); + assert.strictEqual(updated?.status, "failed"); + assert.strictEqual(callCount, 0, "legacy stale batch should not be replayed"); + assert.match(updated?.errors?.[0]?.message ?? "", /Cannot safely recover stale batch/); }); test("processPendingBatches should respect BATCH_MAX_CONCURRENT (default 1)", async () => { diff --git a/tests/unit/batch_api.test.ts b/tests/unit/batch_api.test.ts index 2016877dcf..357e084bf6 100644 --- a/tests/unit/batch_api.test.ts +++ b/tests/unit/batch_api.test.ts @@ -21,6 +21,8 @@ const { formatFileResponse, deleteFile, getTerminalBatches, + ensureBatchItemCheckpoints, + markBatchItemResult, } = await import("../../src/lib/localDb.ts"); const { getDbInstance } = await import("../../src/lib/db/core.ts"); const { @@ -45,6 +47,7 @@ test.afterEach(async () => { } try { const db = getDbInstance(); + db.prepare("DELETE FROM batch_item_checkpoints").run(); db.prepare("DELETE FROM batches").run(); db.prepare("DELETE FROM files").run(); db.prepare("DELETE FROM provider_connections").run(); @@ -695,28 +698,45 @@ test("Batch processor recovers orphaned finalizing batches during startup recove updateBatch(batch.id, { status: "finalizing", finalizingAt: Math.floor(Date.now() / 1000), + requestCountsTotal: 1, + requestCountsCompleted: 1, }); + ensureBatchItemCheckpoints(batch.id, [{ lineNumber: 1, customId: "req-recovery" }]); + markBatchItemResult( + batch.id, + { lineNumber: 1, customId: "req-recovery" }, + { + id: "req_checkpointed_recovery", + custom_id: "req-recovery", + response: { + status_code: 200, + body: { + id: "chatcmpl-mock-recovery", + object: "chat.completion", + model: "gpt-4o-mini", + choices: [ + { + index: 0, + message: { role: "assistant", content: "recovered ok" }, + finish_reason: "stop", + }, + ], + usage: { + prompt_tokens: 1, + completion_tokens: 1, + total_tokens: 2, + }, + }, + }, + } + ); const originalFetch = globalThis.fetch; - (globalThis as any).fetch = async () => { - return Response.json({ - id: "chatcmpl-mock-recovery", - object: "chat.completion", - model: "gpt-4o-mini", - choices: [ - { - index: 0, - message: { role: "assistant", content: "recovered ok" }, - finish_reason: "stop", - }, - ], - usage: { - prompt_tokens: 1, - completion_tokens: 1, - total_tokens: 2, - }, - }); - }; + let fetchCount = 0; + globalThis.fetch = (async () => { + fetchCount++; + throw new Error("checkpointed finalizing batch should not dispatch"); + }) as typeof fetch; try { await processPendingBatches(); @@ -724,6 +744,7 @@ test("Batch processor recovers orphaned finalizing batches during startup recove const recoveredBatch = getBatch(batch.id); assert.strictEqual(recoveredBatch?.status, "completed"); + assert.strictEqual(fetchCount, 0); } finally { globalThis.fetch = originalFetch; }