import { batch } from "./db"; /** Consecutive failed flushes after which a batch is split to find the rows D1 rejects on their own. */ export const ISOLATE_AFTER = 3; const MAX_BACKOFF_MS = 60 * 60 * 1000; export function backoffMs(attempt: number, baseMs: number): number { return Math.min(MAX_BACKOFF_MS, baseMs * 2 ** Math.max(0, attempt - 1)); } export type Poison = { item: T; error: string }; /** * Flushes items as one batch (D1 unless `runBatch` routes it elsewhere, e.g. an account's object). Once `isolate` is set (the batch has failed repeatedly), items are retried * one by one and those that still fail are returned as poison so the caller can dead-letter them instead of * blocking every later flush. If several items all fail individually D1 itself is the problem, so it throws. */ export async function flushIsolating( items: T[], toStatements: (item: T) => S[], isolate: boolean, runBatch: (statements: S[]) => Promise = (statements) => batch(statements as D1PreparedStatement[]), ): Promise[]> { if (items.length === 0) return []; try { await runBatch(items.flatMap(toStatements)); return []; } catch (error) { if (!isolate) throw error; } const poison: Poison[] = []; for (const item of items) { try { await runBatch(toStatements(item)); } catch (error) { poison.push({ item, error: String(error) }); } } if (items.length > 1 && poison.length === items.length) throw new Error(`All ${items.length} rows failed to flush`); return poison; }