From bfcedc617070ef31ba980ab4f2ce2915627848fe Mon Sep 17 00:00:00 2001 From: Rhys Sullivan Date: Thu, 11 Jun 2026 21:12:10 -0700 Subject: [PATCH] Remove the D1-to-R2 oversized-value offload MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The transparent D1 driver wrap (intercept bound params >800KB, stash in R2 under a content hash, rehydrate pointer strings on read) existed only because resolved OpenAPI specs were inlined in integration.config and D1 caps a value at ~1-2MB. Specs now live in the blob seam, written straight to R2 through the handle's blobs backend, so nothing that flows through D1 approaches the cap: across the whole fleet the largest non-spec values are tool rows ~165KB, definitions ~256KB, and plugin_storage ~206KB. The canonical deployment's database was scanned for offload pointer strings before removal — zero exist in any table, so this is code-only. A deployment that still had legacy oversized rows stored as pointers would need the spec backfill before taking this change. The workerd e2e still pushes a ~1MB spec through the real worker; it now proves the spec lands in the R2 blob seam rather than the offload. maxBoundParameters stays — that is D1's parameter-count platform limit, unrelated to value size. --- apps/host-cloudflare/src/db/d1.ts | 15 +- .../src/db/r2-blob-offload.test.ts | 126 ---------- .../host-cloudflare/src/db/r2-blob-offload.ts | 234 ------------------ .../src/worker.e2e.node.test.ts | 15 +- apps/host-cloudflare/wrangler.jsonc | 7 +- 5 files changed, 15 insertions(+), 382 deletions(-) delete mode 100644 apps/host-cloudflare/src/db/r2-blob-offload.test.ts delete mode 100644 apps/host-cloudflare/src/db/r2-blob-offload.ts diff --git a/apps/host-cloudflare/src/db/d1.ts b/apps/host-cloudflare/src/db/d1.ts index dae1b9f24..5402cba3d 100644 --- a/apps/host-cloudflare/src/db/d1.ts +++ b/apps/host-cloudflare/src/db/d1.ts @@ -5,8 +5,6 @@ import { } from "@executor-js/fumadb/adapters/drizzle"; import type { D1Database, R2Bucket } from "@cloudflare/workers-types"; -import { wrapD1WithR2Offload } from "./r2-blob-offload"; - import { collectTables, createExecutorFumaDb, @@ -38,12 +36,8 @@ export const createD1ExecutorDb = async ( provider: "sqlite" as const, }; - // Offload oversized values to R2 (D1 caps a value at ~1-2MB). No-op for - // ordinary small rows; only multi-MB values (e.g. a large OpenAPI spec) leave - // D1. Without a bucket bound, fall back to plain D1 (small values only). - const connection = blobs ? wrapD1WithR2Offload(db, blobs) : db; const schema = createDrizzleRuntimeSchemaFromTables(options); - const drizzleDb = drizzle(connection, { schema }); + const drizzleDb = drizzle(db, { schema }); // D1 rejects SQL `BEGIN TRANSACTION` / `SAVEPOINT` (it requires the JS batch // API), and the shared ensure wraps its DDL in a transaction when the handle @@ -70,9 +64,10 @@ export const createD1ExecutorDb = async ( fuma, // The D1 binding owns its own lifecycle; nothing to release. close: async () => {}, - // Blob-seam writes go straight to R2 — they never enter D1, so they never - // need the offload wrap above (which remains only for legacy oversized - // values already inlined in D1 rows). + // Multi-MB values (resolved OpenAPI specs, introspection snapshots) go + // through the blob seam straight to R2 — they never enter D1, which caps + // a value at ~1-2MB. Without a bucket bound, the executor falls back to + // the FumaDB blob table (small values only). blobs: blobs ? makeR2BlobStore(blobs) : undefined, }; }; diff --git a/apps/host-cloudflare/src/db/r2-blob-offload.test.ts b/apps/host-cloudflare/src/db/r2-blob-offload.test.ts deleted file mode 100644 index fd4e55f09..000000000 --- a/apps/host-cloudflare/src/db/r2-blob-offload.test.ts +++ /dev/null @@ -1,126 +0,0 @@ -import { describe, expect, it } from "@effect/vitest"; -import type { D1Database, R2Bucket } from "@cloudflare/workers-types"; - -import { wrapD1WithR2Offload } from "./r2-blob-offload"; - -// --------------------------------------------------------------------------- -// Round-trip tests for the D1 -> R2 large-value offload. Minimal in-memory -// mocks for the D1 binding (captures the params actually bound; returns canned -// rows on read) and R2 (a Map). Exercises the public D1 surface the wrapper -// presents to drizzle: prepare -> bind -> run/all. -// --------------------------------------------------------------------------- - -const makeMemR2 = () => { - const store = new Map(); - const bucket = { - put: async (key: string, value: ArrayBuffer | ArrayBufferView | string) => { - const bytes = - typeof value === "string" - ? new TextEncoder().encode(value) - : value instanceof ArrayBuffer - ? new Uint8Array(value) - : new Uint8Array(value.buffer, value.byteOffset, value.byteLength); - store.set(key, bytes); - }, - get: async (key: string) => { - const bytes = store.get(key); - if (!bytes) return null; - return { - arrayBuffer: async () => - bytes.buffer.slice(bytes.byteOffset, bytes.byteOffset + bytes.byteLength), - text: async () => new TextDecoder().decode(bytes), - }; - }, - }; - // oxlint-disable-next-line executor/no-double-cast -- test mock: in-memory stand-in for the R2 binding - return { bucket: bucket as unknown as R2Bucket, store }; -}; - -// A fake D1 that records the params bound to the most recent statement and, on -// read, returns whatever rows the test stages. -const makeMockD1 = () => { - const state: { boundParams: unknown[]; rows: Record[] } = { - boundParams: [], - rows: [], - }; - const db = { - prepare: (_sql: string) => { - const stmt: Record = { - bind: (...params: unknown[]) => { - state.boundParams = params; - return stmt; - }, - run: async () => ({ success: true, meta: {}, results: state.rows }), - all: async () => ({ success: true, meta: {}, results: state.rows }), - first: async () => state.rows[0] ?? null, - raw: async () => state.rows.map((r) => Object.values(r)), - }; - return stmt; - }, - batch: async () => [], - exec: async () => ({ count: 0, duration: 0 }), - dump: async () => new ArrayBuffer(0), - }; - // oxlint-disable-next-line executor/no-double-cast -- test mock: in-memory stand-in for the D1 binding - return { db: db as unknown as D1Database, state }; -}; - -const big = "x".repeat(1_000_000); // > 800KB byte threshold - -describe("wrapD1WithR2Offload", () => { - it("offloads an oversized string param to R2 and binds a short pointer", async () => { - const r2 = makeMemR2(); - const mock = makeMockD1(); - const wrapped = wrapD1WithR2Offload(mock.db, r2.bucket); - - await wrapped.prepare("insert into t (a, b) values (?, ?)").bind("small", big).run(); - - expect(mock.state.boundParams[0]).toBe("small"); // small value untouched - const pointer = mock.state.boundParams[1]; - expect(typeof pointer).toBe("string"); - expect(pointer).not.toBe(big); - expect(String(pointer).length).toBeLessThan(200); // a short pointer, not 1MB - expect(r2.store.size).toBe(1); // exactly one blob written - }); - - it("leaves small params inline (no R2 write)", async () => { - const r2 = makeMemR2(); - const mock = makeMockD1(); - const wrapped = wrapD1WithR2Offload(mock.db, r2.bucket); - - await wrapped.prepare("insert into t (a) values (?)").bind("just small").run(); - - expect(mock.state.boundParams).toEqual(["just small"]); - expect(r2.store.size).toBe(0); - }); - - it("rehydrates a pointer back to the original value on read", async () => { - const r2 = makeMemR2(); - const mock = makeMockD1(); - const wrapped = wrapD1WithR2Offload(mock.db, r2.bucket); - - // Write to populate R2 + capture the pointer the column would store. - await wrapped.prepare("insert into t (b) values (?)").bind(big).run(); - const pointer = mock.state.boundParams[0]; - - // Now a read returns that pointer in the row; the wrapper must restore `big`. - mock.state.rows = [{ b: pointer }]; - const result = await wrapped.prepare("select b from t").all(); - - expect(result.results[0]!.b).toBe(big); - }); - - it("fails loud when an offloaded blob is missing (no silent corruption)", async () => { - const r2 = makeMemR2(); - const mock = makeMockD1(); - const wrapped = wrapD1WithR2Offload(mock.db, r2.bucket); - - // Write to mint a real pointer, then simulate R2 losing the object. - await wrapped.prepare("insert into t (b) values (?)").bind(big).run(); - const pointer = mock.state.boundParams[0]; - r2.store.clear(); - - mock.state.rows = [{ b: pointer }]; - await expect(wrapped.prepare("select b from t").all()).rejects.toThrow(/R2 blob lost/); - }); -}); diff --git a/apps/host-cloudflare/src/db/r2-blob-offload.ts b/apps/host-cloudflare/src/db/r2-blob-offload.ts deleted file mode 100644 index 262e0bffd..000000000 --- a/apps/host-cloudflare/src/db/r2-blob-offload.ts +++ /dev/null @@ -1,234 +0,0 @@ -import type { - D1Database, - D1PreparedStatement, - D1Result, - R2Bucket, -} from "@cloudflare/workers-types"; - -// --------------------------------------------------------------------------- -// R2 large-value offload for D1 — transparent, plugin-agnostic, CF-host-only. -// -// D1 caps a single string/BLOB value at ~1-2MB (SQLITE_TOOBIG). Some plugin -// writes are much larger — e.g. the OpenAPI plugin inlines an entire resolved -// spec (Vercel's is ~7MB) into one `plugin_storage.data` row. This wraps the D1 -// binding so any bound parameter over a byte threshold is written to R2 and -// replaced in D1 with a tiny pointer string; on read, pointers are rehydrated -// back to the original value BEFORE drizzle/fumadb sees them (so the JSON-column -// decoder gets the real JSON, never the pointer). -// -// It sits at the D1 driver boundary (below drizzle), so it is column-agnostic -// and needs no schema knowledge: it only ever sees already-serialized string -// values. Keyed by content hash (idempotent + dedup). Deletes intentionally -// leave the R2 object (orphans are cheap; a sweeper is a future improvement). -// -// The proper long-term fix is plugin-level (store large specs via the executor -// `blobs`/BlobStore seam) — see the TODO in packages/plugins/openapi. -// --------------------------------------------------------------------------- - -// Astronomically-unlikely sentinel: a real value would have to be short AND -// start with this exact magic to be falsely rehydrated. -const POINTER_PREFIX = "\u0000__executor_r2_blob_v1__:"; - -// Offload values larger than this many UTF-8 bytes. Well under D1's ~1MB cap, -// and large enough that ordinary plugin rows (tiny) never touch R2. -const OFFLOAD_BYTE_THRESHOLD = 800_000; -// Cheap pre-filter: only measure byte length for strings longer than this many -// UTF-16 units (skips the millions of tiny params without a TextEncoder pass). -const LENGTH_PREFILTER = 200_000; - -const encoder = new TextEncoder(); - -const toHex = (buffer: ArrayBuffer): string => { - const bytes = new Uint8Array(buffer); - let out = ""; - for (const b of bytes) out += b.toString(16).padStart(2, "0"); - return out; -}; - -const hashKey = async (bytes: Uint8Array): Promise => - // oxlint-disable-next-line executor/no-double-cast -- boundary: Workers vs DOM BufferSource type mismatch for crypto.subtle.digest - `blobs/${toHex(await crypto.subtle.digest("SHA-256", bytes as unknown as BufferSource))}`; - -// A bound param's storage form. fumadb encodes a `json` column to UTF-8 BYTES -// (bound as a D1 BLOB), so the oversized value is usually a Uint8Array — not a -// string or object. We offload by storage kind so the read can return the SAME -// shape D1 natively would (a BLOB cell -> ArrayBuffer, a TEXT cell -> string), -// keeping drizzle's column decoder happy. -type Offload = { kind: "b" | "t"; bytes: Uint8Array }; - -const oversizedOffload = (value: unknown): Offload | null => { - if (typeof value === "string") { - if (value.length <= LENGTH_PREFILTER) return null; - const bytes = encoder.encode(value); - return bytes.byteLength > OFFLOAD_BYTE_THRESHOLD ? { kind: "t", bytes } : null; - } - if (value instanceof ArrayBuffer) { - return value.byteLength > OFFLOAD_BYTE_THRESHOLD - ? { kind: "b", bytes: new Uint8Array(value) } - : null; - } - if (ArrayBuffer.isView(value)) { - return value.byteLength > OFFLOAD_BYTE_THRESHOLD - ? { kind: "b", bytes: new Uint8Array(value.buffer, value.byteOffset, value.byteLength) } - : null; - } - if (value !== null && typeof value === "object") { - const bytes = encoder.encode(JSON.stringify(value)); - return bytes.byteLength > OFFLOAD_BYTE_THRESHOLD ? { kind: "t", bytes } : null; - } - return null; -}; - -const isPointer = (value: unknown): value is string => - typeof value === "string" && value.startsWith(POINTER_PREFIX); - -// Minimal retry for transient R2 failures (network blips). 3 attempts with -// small exponential backoff; the final failure propagates with context so a -// real outage surfaces rather than corrupting a read/write. -const withRetry = async (op: () => Promise, what: string): Promise => { - let lastError: unknown; - for (let attempt = 0; attempt < 3; attempt++) { - // oxlint-disable-next-line executor/no-try-catch-or-throw -- boundary: retry transient R2 I/O before failing loud - try { - return await op(); - } catch (error) { - lastError = error; - if (attempt < 2) await new Promise((resolve) => setTimeout(resolve, 20 * 2 ** attempt)); - } - } - // oxlint-disable-next-line executor/no-error-constructor, executor/no-try-catch-or-throw -- boundary: D1/R2 driver wrapper (not Effect domain); surface the exhausted R2 failure with context - throw new Error(`Failed to ${what} after 3 attempts`, { cause: lastError }); -}; - -/** - * Write any oversized params to R2; replace each with a pointer string - * (`:`). The pointer is a short TEXT value D1 stores happily; - * the read path rehydrates it back to the original BLOB/TEXT shape before - * drizzle's column decoder runs. - */ -const offloadParams = (params: unknown[], bucket: R2Bucket): Promise => - Promise.all( - params.map(async (param) => { - const offload = oversizedOffload(param); - if (!offload) return param; - const key = await hashKey(offload.bytes); - await withRetry(() => bucket.put(key, offload.bytes), `offload value to R2 (${key})`); - return `${POINTER_PREFIX}${offload.kind}:${key}`; - }), - ); - -/** Resolve a single value: rehydrate from R2 if it is a pointer, else pass through. */ -const rehydrateValue = async (value: unknown, bucket: R2Bucket): Promise => { - if (!isPointer(value)) return value; - const rest = value.slice(POINTER_PREFIX.length); - const kind = rest[0]; - const key = rest.slice(2); // skip ":" - const object = await withRetry(() => bucket.get(key), `read offloaded value from R2 (${key})`); - // Fail LOUD on a lost blob — returning the raw pointer here would feed garbage - // into drizzle's column decoder and silently corrupt the read. - // oxlint-disable-next-line executor/no-error-constructor, executor/no-try-catch-or-throw -- boundary: D1/R2 driver wrapper (not Effect domain); R2 durability failure must surface, not silently pass the pointer through - if (!object) throw new Error(`R2 blob lost for offloaded D1 value: ${key}`); - // Return the SAME shape D1 would for that storage class: BLOB -> ArrayBuffer, - // TEXT -> string. (The json column is a BLOB, so fumadb gets bytes to decode.) - return kind === "b" ? await object.arrayBuffer() : await object.text(); -}; - -const rehydrateRow = async ( - row: Record, - bucket: R2Bucket, -): Promise> => { - let out: Record | null = null; - for (const [col, value] of Object.entries(row)) { - if (isPointer(value)) { - out ??= { ...row }; - out[col] = await rehydrateValue(value, bucket); - } - } - return out ?? row; -}; - -const rehydrateRows = ( - rows: T[] | undefined, - bucket: R2Bucket, -): Promise | T[] | undefined => { - if (!rows || rows.length === 0) return rows; - return Promise.all( - rows.map((row) => - row && typeof row === "object" - ? (rehydrateRow(row as Record, bucket) as Promise) - : Promise.resolve(row), - ), - ); -}; - -const rehydrateResult = async (result: T, bucket: R2Bucket): Promise => { - const rehydrated = await rehydrateRows(result.results as unknown[] | undefined, bucket); - return rehydrated === result.results ? result : { ...result, results: rehydrated }; -}; - -/** - * Wrap a `D1Database` so oversized bound values offload to R2 transparently. - * Returns an object satisfying the `D1Database` surface drizzle-d1 uses - * (`prepare` + the prepared-statement run/all/first/raw/bind methods, plus - * `batch`/`exec`/`dump` pass-through). Pure delegation otherwise. - */ -export const wrapD1WithR2Offload = (db: D1Database, bucket: R2Bucket): D1Database => { - const wrapStatement = ( - statement: D1PreparedStatement, - params: unknown[] | null, - ): D1PreparedStatement => { - // Bind happens synchronously in the D1 API, but offload is async — so we - // capture the params and only bind the (possibly offloaded) values when the - // terminal run/all/first/raw runs. - const bound = async (): Promise => - params ? statement.bind(...(await offloadParams(params, bucket))) : statement; - - const wrapped: D1PreparedStatement = { - bind: (...next: unknown[]) => wrapStatement(statement, next), - first: (async (colName?: string) => { - const s = await bound(); - const value = colName === undefined ? await s.first() : await s.first(colName); - if (value === null || value === undefined) return value; - if (colName !== undefined) return rehydrateValue(value, bucket); - return rehydrateRow(value as Record, bucket); - }) as D1PreparedStatement["first"], - run: (async () => - rehydrateResult(await (await bound()).run(), bucket)) as D1PreparedStatement["run"], - all: (async () => - rehydrateResult(await (await bound()).all(), bucket)) as D1PreparedStatement["all"], - raw: (async (options?: { columnNames?: boolean }) => { - // Call `.raw()` directly on the statement (do NOT extract the method — - // the D1 runtime reads `this.statement`, so a detached call throws). - // oxlint-disable-next-line executor/no-double-cast -- boundary: collapse D1 raw() overloads to one callable shape - const s = (await bound()) as unknown as { - raw(o?: { columnNames?: boolean }): Promise; - }; - const rows = await s.raw(options); - // `raw()` returns arrays of column values (optionally a header row of - // column names first); rehydrate the value cells, leave any header row. - return Promise.all( - rows.map((row, index) => - Array.isArray(row) && !(options?.columnNames && index === 0) - ? Promise.all(row.map((cell) => rehydrateValue(cell, bucket))) - : Promise.resolve(row), - ), - ); - }) as D1PreparedStatement["raw"], - }; - return wrapped; - }; - - // Intercept only `prepare` (to wrap statements); delegate everything else - // (batch/exec/dump/withSession + any internal fields) to the real binding, - // binding methods so D1's internal `this` stays intact. A Proxy preserves the - // D1Database type without enumerating-and-casting the surface. - return new Proxy(db, { - get(target, prop, receiver) { - if (prop === "prepare") { - return (query: string) => wrapStatement(target.prepare(query), null); - } - const value = Reflect.get(target, prop, receiver); - return typeof value === "function" ? value.bind(target) : value; - }, - }); -}; diff --git a/apps/host-cloudflare/src/worker.e2e.node.test.ts b/apps/host-cloudflare/src/worker.e2e.node.test.ts index 37f2c1176..63287f461 100644 --- a/apps/host-cloudflare/src/worker.e2e.node.test.ts +++ b/apps/host-cloudflare/src/worker.e2e.node.test.ts @@ -10,8 +10,8 @@ import { unstable_dev, type Unstable_DevWorker } from "wrangler"; // End-to-end test for the Cloudflare host: boots the REAL worker on workerd via // Miniflare (wrangler `unstable_dev`) with a local D1 + R2, dev-auth on. This is // the only test that exercises the CF-specific stack together — D1 schema -// bring-up, the R2 large-value offload, QuickJS-WASM execution, and the MCP -// envelope — through the actual HTTP surface. +// bring-up, the R2-backed blob seam (multi-MB spec storage), QuickJS-WASM +// execution, and the MCP envelope — through the actual HTTP surface. // --------------------------------------------------------------------------- const dir = fileURLToPath(new URL(".", import.meta.url)); @@ -77,11 +77,10 @@ describe("cloudflare host e2e (workerd/miniflare)", () => { expect(body.text).toBe("42"); }, 60_000); - it("adds a LARGE OpenAPI source — exercises R2 offload (>800KB blob) + createMany batching (>100 tools)", async () => { - // Synthesize a spec big enough to (a) push the stored config blob past the - // ~800KB R2-offload threshold and (b) derive >100 tools (past D1's 100 - // bound-param createMany limit) — the real-worker regression for two of the - // three D1 fixes. + it("adds a LARGE OpenAPI source — exercises the R2 blob seam (~1MB spec) + createMany batching (>100 tools)", async () => { + // Synthesize a spec big enough to (a) far exceed D1's per-value cap if it + // were inlined — proving the spec text really lands in the R2 blob seam — + // and (b) derive >100 tools (past D1's 100 bound-param createMany limit). const paths: Record = {}; for (let i = 0; i < 250; i++) { paths[`/op${i}`] = { @@ -116,7 +115,7 @@ describe("cloudflare host e2e (workerd/miniflare)", () => { const added = (await add.json()) as { toolCount: number }; expect(added.toolCount).toBe(250); - // Reads back through the R2 rehydration path (the >800KB blob lives in R2). + // Reads back the catalog row whose config points at the R2 spec blob. const got = await worker.fetch(`/api/openapi/integrations/${slug}`); expect(got.status).toBe(200); const integration = (await got.json()) as { slug: string } | null; diff --git a/apps/host-cloudflare/wrangler.jsonc b/apps/host-cloudflare/wrangler.jsonc index 4e9aba169..aec20421f 100644 --- a/apps/host-cloudflare/wrangler.jsonc +++ b/apps/host-cloudflare/wrangler.jsonc @@ -24,10 +24,9 @@ "database_id": "ae748ca1-032c-4427-a1a0-fe39db77d1a9", }, ], - // R2 holds oversized values that exceed D1's per-value cap (~1-2MB). The D1 - // handle offloads large bound params to this bucket and stores a pointer in - // the row (apps/host-cloudflare/src/db/r2-blob-offload.ts). `wrangler r2 - // bucket create executor-blobs` provisions it. + // Plugin blob seam backend: multi-MB values (resolved OpenAPI specs, + // introspection snapshots) live here, not in D1 (which caps a value at + // ~1-2MB). `wrangler r2 bucket create executor-blobs` provisions it. "r2_buckets": [ { "binding": "BLOBS",