diff --git a/apps/host-cloudflare/src/db/d1-retry.test.ts b/apps/host-cloudflare/src/db/d1-retry.test.ts new file mode 100644 index 0000000000..c1fb5f5573 --- /dev/null +++ b/apps/host-cloudflare/src/db/d1-retry.test.ts @@ -0,0 +1,270 @@ +import { describe, expect, it } from "@effect/vitest"; +import { Duration, Effect, Exit, Fiber } from "effect"; +import { TestClock } from "effect/testing"; +import type { D1Database, D1PreparedStatement } from "@cloudflare/workers-types"; + +import { + D1CallError, + classifyD1Failure, + isReadOnlySql, + retryTransientD1Call, + withD1TransientRetry, +} from "./d1-retry"; + +const bindingRejection = (message: string): Error => + // oxlint-disable-next-line executor/no-error-constructor -- test double: the D1 binding rejects with a plain Error carrying the documented message + new Error(message); + +const INCIDENT_ERROR = "D1_ERROR: internal error; reference = 6kq3v9j0mmpdu3qgcsbmz2ib"; +const UNREACHABLE_ERROR = "D1_ERROR: Cannot resolve D1 DB due to transient issue on remote node."; +const CONSTRAINT_ERROR = "D1_ERROR: UNIQUE constraint failed: integration.slug: SQLITE_CONSTRAINT"; + +const failingCall = (failures: readonly string[], success: string) => { + const calls = { count: 0 }; + const call = (): Promise => { + calls.count += 1; + const failure = failures[calls.count - 1]; + return failure === undefined + ? Promise.resolve(success) + : // oxlint-disable-next-line executor/no-promise-reject -- test double: a D1 call that rejects like the binding does + Promise.reject(bindingRejection(failure)); + }; + return { call, calls }; +}; + +const runWithVirtualTime = (effect: Effect.Effect) => + Effect.gen(function* () { + const fiber = yield* Effect.forkChild(Effect.exit(effect)); + yield* TestClock.adjust(Duration.seconds(30)); + return yield* Fiber.join(fiber); + }); + +describe("classifyD1Failure", () => { + it.each([ + [INCIDENT_ERROR, "maybe-executed"], + ["D1_ERROR: Network connection lost.", "maybe-executed"], + ["D1_ERROR: Replica disconnected from primary.", "maybe-executed"], + ["D1_ERROR: Internal error in D1 DB storage caused object to be reset.", "maybe-executed"], + ["D1_ERROR: D1 DB reset because its code was updated.", "maybe-executed"], + ["D1_ERROR: Can't read from request stream because client disconnected.", "maybe-executed"], + ["D1_ERROR: D1 DB is overloaded. Requests queued for too long.", "maybe-executed"], + [UNREACHABLE_ERROR, "not-executed"], + [ + "D1_ERROR: Internal error while starting up D1 DB storage caused object to be reset.", + "not-executed", + ], + ["D1_ERROR: D1 DB is overloaded. Too many requests queued.", "not-executed"], + [CONSTRAINT_ERROR, "permanent"], + ['D1_ERROR: near "SELEC": syntax error at offset 0: SQLITE_ERROR', "permanent"], + ["D1_ERROR: no such table: integration: SQLITE_ERROR", "permanent"], + ["D1_ERROR: No SQL statements detected.", "permanent"], + ["D1_ERROR: Exceeded maximum DB size.", "permanent"], + [ + "D1_ERROR: D1 DB storage operation exceeded timeout which caused object to be reset.", + "permanent", + ], + ["D1_ERROR: D1 DB's isolate exceeded its memory limit and was reset.", "permanent"], + ["D1_ERROR: D1 DB exceeded its CPU time limit and was reset.", "permanent"], + ["D1_TYPE_ERROR: Type 'undefined' not supported for value 'undefined'", "permanent"], + ["internal error", "permanent"], + ])("%s -> %s", (message, expected) => { + expect(classifyD1Failure(bindingRejection(message))).toBe(expected); + }); + + it("reads the documented message from the error cause", () => { + // oxlint-disable-next-line executor/no-error-constructor -- test double: drizzle wraps the binding error as the cause of its own query error + const wrapped = new Error("Failed query: select 1", { + cause: bindingRejection(INCIDENT_ERROR), + }); + expect(classifyD1Failure(wrapped)).toBe("maybe-executed"); + }); + + it("treats a non-error rejection as permanent", () => { + expect(classifyD1Failure("D1_ERROR: internal error")).toBe("permanent"); + }); +}); + +describe("isReadOnlySql", () => { + it.each([ + ['select "id" from "integration" where "tenant" = ?', true], + [" WITH t AS (SELECT 1) SELECT * FROM t", true], + ["EXPLAIN QUERY PLAN SELECT 1", true], + ['insert into "integration" ("id") values (?)', false], + ['update "integration" set "name" = ?', false], + ["WITH t AS (SELECT 1) DELETE FROM integration", false], + ["CREATE TABLE IF NOT EXISTS t (id text)", false], + ["PRAGMA table_info('connection')", false], + ])("%s -> %s", (sql, expected) => { + expect(isReadOnlySql(sql)).toBe(expected); + }); +}); + +describe("retryTransientD1Call", () => { + it.effect("succeeds on the fifth attempt after four transient failures of a read", () => + Effect.gen(function* () { + const { call, calls } = failingCall( + [INCIDENT_ERROR, INCIDENT_ERROR, INCIDENT_ERROR, INCIDENT_ERROR], + "rows", + ); + + const exit = yield* runWithVirtualTime(retryTransientD1Call("all", true, call)); + + expect(exit).toEqual(Exit.succeed("rows")); + expect(calls.count).toBe(5); + }), + ); + + it.effect("gives up after five attempts and surfaces the original error", () => + Effect.gen(function* () { + const { call, calls } = failingCall( + [INCIDENT_ERROR, INCIDENT_ERROR, INCIDENT_ERROR, INCIDENT_ERROR, INCIDENT_ERROR], + "rows", + ); + + const exit = yield* runWithVirtualTime(retryTransientD1Call("all", true, call)); + + expect(exit).toEqual(Exit.fail(new D1CallError({ cause: bindingRejection(INCIDENT_ERROR) }))); + expect(calls.count).toBe(5); + }), + ); + + it.effect("does not retry a constraint violation", () => + Effect.gen(function* () { + const { call, calls } = failingCall([CONSTRAINT_ERROR], "ok"); + + const exit = yield* runWithVirtualTime(retryTransientD1Call("run", false, call)); + + expect(exit).toEqual( + Exit.fail(new D1CallError({ cause: bindingRejection(CONSTRAINT_ERROR) })), + ); + expect(calls.count).toBe(1); + }), + ); + + it.effect("does not retry a write after a failure that may have applied it", () => + Effect.gen(function* () { + const { call, calls } = failingCall([INCIDENT_ERROR], "ok"); + + const exit = yield* runWithVirtualTime(retryTransientD1Call("run", false, call)); + + expect(exit).toEqual(Exit.fail(new D1CallError({ cause: bindingRejection(INCIDENT_ERROR) }))); + expect(calls.count).toBe(1); + }), + ); + + it.effect("retries a write when the database was never reached", () => + Effect.gen(function* () { + const { call, calls } = failingCall([UNREACHABLE_ERROR, UNREACHABLE_ERROR], "ok"); + + const exit = yield* runWithVirtualTime(retryTransientD1Call("run", false, call)); + + expect(exit).toEqual(Exit.succeed("ok")); + expect(calls.count).toBe(3); + }), + ); + + it.effect("waits between attempts instead of retrying immediately", () => + Effect.gen(function* () { + const { call, calls } = failingCall([INCIDENT_ERROR], "rows"); + + const fiber = yield* Effect.forkChild(retryTransientD1Call("all", true, call)); + yield* TestClock.adjust(Duration.millis(100)); + const attemptsBeforeFirstDelay = calls.count; + yield* TestClock.adjust(Duration.millis(200)); + const result = yield* Fiber.join(fiber); + + expect(attemptsBeforeFirstDelay).toBe(1); + expect(result).toBe("rows"); + expect(calls.count).toBe(2); + }), + ); +}); + +const makeFakeD1 = (failures: readonly string[]) => { + const executed: string[] = []; + const batches: unknown[][] = []; + let remainingFailures = [...failures]; + const settle = (sql: string, value: A): Promise => { + executed.push(sql); + const [failure, ...rest] = remainingFailures; + remainingFailures = rest; + return failure === undefined + ? Promise.resolve(value) + : // oxlint-disable-next-line executor/no-promise-reject -- test double: a D1 call that rejects like the binding does + Promise.reject(bindingRejection(failure)); + }; + const created: D1PreparedStatement[] = []; + const statement = (sql: string, args: readonly unknown[]): D1PreparedStatement => { + // oxlint-disable-next-line executor/no-double-cast -- test double: D1 result metadata is irrelevant to the retry wrapper + const made = { + bind: (...values: readonly unknown[]) => statement(sql, [...args, ...values]), + first: () => settle(sql, null), + run: () => settle(sql, { success: true, meta: {}, results: [] }), + all: () => settle(sql, { success: true, meta: {}, results: [{ args }] }), + raw: () => settle(sql, []), + } as unknown as D1PreparedStatement; + created.push(made); + return made; + }; + const prepare = (sql: string) => statement(sql, []); + const batch = (statements: D1PreparedStatement[]) => { + batches.push(statements); + return settle("batch", []); + }; + // oxlint-disable-next-line executor/no-double-cast -- test double: only the D1 methods the wrapper forwards are implemented + const db = { + prepare, + batch, + withSession: () => ({ prepare, batch, getBookmark: () => null }), + } as unknown as D1Database; + return { db, executed, batches, created }; +}; + +describe("withD1TransientRetry", () => { + it.live("retries a bound read through the binding and returns its rows", () => + Effect.gen(function* () { + const fake = makeFakeD1([INCIDENT_ERROR]); + const db = withD1TransientRetry(fake.db); + + const result = yield* Effect.promise(() => + db.withSession("first-primary").prepare("SELECT 1 WHERE ? = ?").bind("a", "b").all(), + ); + + expect(result.results).toEqual([{ args: ["a", "b"] }]); + expect(fake.executed).toEqual(["SELECT 1 WHERE ? = ?", "SELECT 1 WHERE ? = ?"]); + }), + ); + + it.live("rejects a write with the original binding error after one attempt", () => + Effect.gen(function* () { + const fake = makeFakeD1([INCIDENT_ERROR]); + const db = withD1TransientRetry(fake.db); + + const rejection = yield* Effect.promise(() => + db + .prepare("INSERT INTO t (id) VALUES (?)") + .bind("x") + .run() + .then( + () => "resolved", + (failure: unknown) => failure, + ), + ); + + expect(rejection).toEqual(bindingRejection(INCIDENT_ERROR)); + expect(fake.executed).toEqual(["INSERT INTO t (id) VALUES (?)"]); + }), + ); + + it.live("hands the binding its own statements in a batch", () => + Effect.gen(function* () { + const fake = makeFakeD1([]); + const db = withD1TransientRetry(fake.db); + const statements = [db.prepare("SELECT 1"), db.prepare("SELECT 2").bind(7)]; + + yield* Effect.promise(() => db.batch(statements)); + + expect(fake.batches).toEqual([[fake.created[0], fake.created[2]]]); + }), + ); +}); diff --git a/apps/host-cloudflare/src/db/d1-retry.ts b/apps/host-cloudflare/src/db/d1-retry.ts new file mode 100644 index 0000000000..acd8c84da2 --- /dev/null +++ b/apps/host-cloudflare/src/db/d1-retry.ts @@ -0,0 +1,198 @@ +import { Data, Duration, Effect, Predicate, Schedule } from "effect"; +import type { D1Database, D1DatabaseSession, D1PreparedStatement } from "@cloudflare/workers-types"; + +// --------------------------------------------------------------------------- +// Transient D1 failure retry. +// +// D1 occasionally fails a burst of queries with infrastructure errors (the +// Durable Object behind the database restarts, a node is briefly unreachable) +// and then recovers on its own. Every executor query goes through the D1 +// binding, so the binding is wrapped once here and each call rides out a +// short blip with a bounded, jittered exponential backoff. +// +// A failure is retried only when Cloudflare documents it as transient +// (https://developers.cloudflare.com/d1/observability/debug-d1/), and how far +// it is retried depends on whether the statement may already have run: +// +// - "not-executed": the request never reached a running database, so the +// statement did not run. Safe to retry any statement, writes included. +// - "maybe-executed": the database failed while (or after) the statement ran. +// A write may or may not have applied, and the binding cannot tell which, +// so only read-only statements are retried. A write surfaces the original +// error, as before. +// - "permanent": everything else (SQL errors, constraint violations, limits, +// query-caused timeouts and resets). Never retried. +// --------------------------------------------------------------------------- + +export type D1FailureClass = "not-executed" | "maybe-executed" | "permanent"; + +const NOT_EXECUTED_MARKERS = [ + "Cannot resolve D1 DB due to transient issue on remote node", + "Internal error while starting up D1 DB storage caused object to be reset", + "D1 DB is overloaded. Too many requests queued", +] as const; + +const MAYBE_EXECUTED_MARKERS = [ + "D1_ERROR: internal error", + "Network connection lost", + "Replica disconnected from primary", + "Internal error in D1 DB storage caused object to be reset", + "D1 DB reset because its code was updated", + "Can't read from request stream because client disconnected", + "D1 DB is overloaded. Requests queued for too long", +] as const; + +const MAX_CAUSE_DEPTH = 4; + +const failureMessages = (failure: unknown): readonly string[] => { + const messages: string[] = []; + let current: unknown = failure; + for (let depth = 0; depth < MAX_CAUSE_DEPTH && Predicate.isError(current); depth += 1) { + // oxlint-disable-next-line executor/no-unknown-error-message -- adapter boundary: the D1 binding rejects with a plain Error whose message IS the documented error code + messages.push(current.message); + current = current.cause; + } + return messages; +}; + +export const classifyD1Failure = (failure: unknown): D1FailureClass => { + const messages = failureMessages(failure); + const mentions = (markers: readonly string[]) => + messages.some((message) => markers.some((marker) => message.includes(marker))); + if (mentions(NOT_EXECUTED_MARKERS)) return "not-executed"; + if (mentions(MAYBE_EXECUTED_MARKERS)) return "maybe-executed"; + return "permanent"; +}; + +const READ_ONLY_LEADING_KEYWORD = /^\s*(?:SELECT|WITH|EXPLAIN)\b/i; +const WRITE_KEYWORD = + /\b(?:INSERT|UPDATE|DELETE|REPLACE|UPSERT|CREATE|DROP|ALTER|PRAGMA|ATTACH|DETACH|VACUUM|REINDEX|ANALYZE|BEGIN|COMMIT|ROLLBACK|SAVEPOINT|RELEASE)\b/i; + +/** Mirrors D1's own read-only detection: only SELECT / WITH / EXPLAIN, and no + * keyword that can write anywhere in the text. A false "write" only means the + * statement is not retried on a maybe-executed failure. */ +export const isReadOnlySql = (sql: string): boolean => + READ_ONLY_LEADING_KEYWORD.test(sql) && !WRITE_KEYWORD.test(sql); + +export const isRetryableD1Failure = (failure: unknown, readOnly: boolean): boolean => { + const failureClass = classifyD1Failure(failure); + if (failureClass === "not-executed") return true; + return failureClass === "maybe-executed" && readOnly; +}; + +export const D1_RETRY_MAX_ATTEMPTS = 5; +const D1_RETRY_BASE_DELAY = Duration.millis(200); +const D1_RETRY_MAX_DELAY = Duration.seconds(2); + +/** 200ms, 400ms, 800ms, 1600ms (each jittered to 80-120%, capped at 2s): + * five attempts within about 3.6 seconds, so an MCP call still answers. */ +export const d1RetrySchedule = Schedule.exponential(D1_RETRY_BASE_DELAY).pipe( + Schedule.modifyDelay((_, delay) => Effect.succeed(Duration.min(delay, D1_RETRY_MAX_DELAY))), + Schedule.jittered, + Schedule.both(Schedule.recurs(D1_RETRY_MAX_ATTEMPTS - 1)), +); + +/** A rejection from the D1 binding, kept as-is in `cause`. */ +export class D1CallError extends Data.TaggedError("D1CallError")<{ + readonly cause: unknown; +}> {} + +export const retryTransientD1Call = ( + operation: string, + readOnly: boolean, + call: () => Promise, +): Effect.Effect => { + let attempt = 0; + return Effect.suspend(() => { + attempt += 1; + const current = attempt; + return Effect.tryPromise({ try: call, catch: (cause) => new D1CallError({ cause }) }).pipe( + Effect.tapError(({ cause: failure }) => + current < D1_RETRY_MAX_ATTEMPTS && isRetryableD1Failure(failure, readOnly) + ? Effect.logWarning("D1 transient failure, retrying").pipe( + Effect.annotateLogs({ + "d1.operation": operation, + "d1.read_only": readOnly, + "d1.attempt": current, + "d1.max_attempts": D1_RETRY_MAX_ATTEMPTS, + "d1.failure_class": classifyD1Failure(failure), + "d1.failure": failureMessages(failure)[0] ?? "unknown", + }), + ) + : Effect.void, + ), + ); + }).pipe( + Effect.retry({ + schedule: d1RetrySchedule, + while: (failure) => isRetryableD1Failure(failure.cause, readOnly), + }), + ); +}; + +const runRetried = (operation: string, readOnly: boolean, call: () => Promise): Promise => + Effect.runPromise( + retryTransientD1Call(operation, readOnly, call).pipe( + Effect.mapError((failure) => failure.cause), + ), + ); + +interface WrappedStatement { + readonly inner: D1PreparedStatement; + readonly readOnly: boolean; +} + +const wrappedStatements = new WeakMap(); + +const wrapStatement = (inner: D1PreparedStatement, sql: string): D1PreparedStatement => { + const readOnly = isReadOnlySql(sql); + const first = (colName?: string): Promise => + runRetried("first", readOnly, () => + colName === undefined ? inner.first() : inner.first(colName), + ); + function raw(options: { columnNames: true }): Promise<[string[], ...T[]]>; + function raw(options?: { columnNames?: false }): Promise; + function raw(options?: { + readonly columnNames?: boolean; + }): Promise<[string[], ...T[]] | T[]> { + return runRetried<[string[], ...T[]] | T[]>("raw", readOnly, () => + options?.columnNames === true ? inner.raw({ columnNames: true }) : inner.raw(), + ); + } + const wrapped: D1PreparedStatement = { + bind: (...values) => wrapStatement(inner.bind(...values), sql), + first, + run: () => runRetried("run", readOnly, () => inner.run()), + all: () => runRetried("all", readOnly, () => inner.all()), + raw, + }; + wrappedStatements.set(wrapped, { inner, readOnly }); + return wrapped; +}; + +/** A batch runs as one transaction, so it may be retried after a + * maybe-executed failure only when every statement in it is read-only. */ +const wrapBatch = + (batch: D1Database["batch"]): D1Database["batch"] => + (statements) => + runRetried( + "batch", + statements.every((statement) => wrappedStatements.get(statement)?.readOnly === true), + () => + batch(statements.map((statement) => wrappedStatements.get(statement)?.inner ?? statement)), + ); + +const wrapSession = (inner: D1DatabaseSession): D1DatabaseSession => ({ + prepare: (query) => wrapStatement(inner.prepare(query), query), + batch: wrapBatch((statements) => inner.batch(statements)), + getBookmark: () => inner.getBookmark(), +}); + +/** The D1 binding with every query retried on documented transient failures. */ +export const withD1TransientRetry = (db: D1Database): D1Database => ({ + prepare: (query) => wrapStatement(db.prepare(query), query), + batch: wrapBatch((statements) => db.batch(statements)), + exec: (query) => runRetried("exec", isReadOnlySql(query), () => db.exec(query)), + withSession: (constraintOrBookmark) => wrapSession(db.withSession(constraintOrBookmark)), + dump: () => db.dump(), +}); diff --git a/apps/host-cloudflare/src/db/d1.test.ts b/apps/host-cloudflare/src/db/d1.test.ts index 5ccf270e56..871f78c01f 100644 --- a/apps/host-cloudflare/src/db/d1.test.ts +++ b/apps/host-cloudflare/src/db/d1.test.ts @@ -11,15 +11,16 @@ const makeRecordingD1 = ( ): { readonly db: D1Database; readonly statements: string[]; - readonly failWhen: (predicate: ((sql: string) => boolean) | null) => void; + readonly failWhen: (predicate: ((sql: string) => boolean) | null, message?: string) => void; } => { const statements: string[] = []; let failurePredicate: ((sql: string) => boolean) | null = null; + let failureMessage = "forced D1 failure"; const record = (sql: string): void => { statements.push(sql); if (failurePredicate?.(sql)) { // oxlint-disable-next-line executor/no-try-catch-or-throw, executor/no-error-constructor -- test boundary: the fake D1 adapter must reject exactly where the real D1 query would reject - throw new Error("forced D1 failure"); + throw new Error(failureMessage); } }; const prepare = (sql: string) => { @@ -47,8 +48,9 @@ const makeRecordingD1 = ( return { db, statements, - failWhen: (predicate) => { + failWhen: (predicate, message = "forced D1 failure") => { failurePredicate = predicate; + failureMessage = message; }, }; }; @@ -204,4 +206,28 @@ describe("createD1ExecutorDb", () => { await sqlite.close(); }); + + it("rides out a transient D1 internal error on the schema fingerprint read", async () => { + const sqlite = await createSqliteTestFumaDb({ tables: collectTables() }); + const { db, statements, failWhen } = makeRecordingD1(sqlite.client); + const prepared = await createD1ExecutorDb(db, undefined); + await prepared.close(); + let fingerprintReads = 0; + failWhen((sql) => { + if (!sql.startsWith("SELECT fingerprint")) return false; + fingerprintReads += 1; + return fingerprintReads === 1; + }, "D1_ERROR: internal error; reference = 6kq3v9j0mmpdu3qgcsbmz2ib"); + + const secondStart = statements.length; + const handle = await createD1ExecutorDb(db, undefined); + await handle.close(); + + expect( + statements.slice(secondStart).filter((sql) => sql.startsWith("SELECT fingerprint")), + ).toHaveLength(2); + expect(statements.slice(secondStart).filter(isRuntimeSchemaStatement)).toEqual([]); + + await sqlite.close(); + }); }); diff --git a/apps/host-cloudflare/src/db/d1.ts b/apps/host-cloudflare/src/db/d1.ts index 43bc375878..5763670987 100644 --- a/apps/host-cloudflare/src/db/d1.ts +++ b/apps/host-cloudflare/src/db/d1.ts @@ -15,6 +15,7 @@ import { makeR2BlobStore } from "@executor-js/cloudflare/blob-store"; import { CLOUDFLARE_NAMESPACE, CLOUDFLARE_SCHEMA_VERSION } from "../config"; import { prepareCloudflareD1Data } from "./data-migrations"; +import { withD1TransientRetry } from "./d1-retry"; const SCHEMA_FINGERPRINT_TABLE = `private_${CLOUDFLARE_NAMESPACE}_schema_fingerprint`; const SCHEMA_FINGERPRINT_ID = "runtime-schema"; @@ -85,9 +86,10 @@ const storePreparedSchemaFingerprint = async ( // --------------------------------------------------------------------------- export const createD1ExecutorDb = async ( - db: D1Database, + binding: D1Database, blobs: R2Bucket | undefined, ): Promise => { + const db = withD1TransientRetry(binding); const options = { tables: collectTables(), namespace: CLOUDFLARE_NAMESPACE,