From eb4ad212f967958491dd96e6f93c09d56f92af60 Mon Sep 17 00:00:00 2001 From: Jad Chalita Date: Thu, 23 Jul 2026 08:42:04 +0200 Subject: [PATCH] Skip redundant Cloudflare D1 schema preparation --- apps/host-cloudflare/src/db/d1.test.ts | 207 ++++++++++++++++++ apps/host-cloudflare/src/db/d1.ts | 89 +++++++- .../host-cloudflare/src/db/data-migrations.ts | 26 ++- 3 files changed, 311 insertions(+), 11 deletions(-) create mode 100644 apps/host-cloudflare/src/db/d1.test.ts diff --git a/apps/host-cloudflare/src/db/d1.test.ts b/apps/host-cloudflare/src/db/d1.test.ts new file mode 100644 index 000000000..5ccf270e5 --- /dev/null +++ b/apps/host-cloudflare/src/db/d1.test.ts @@ -0,0 +1,207 @@ +import { describe, expect, it } from "@effect/vitest"; +import type { D1Database } from "@cloudflare/workers-types"; + +import { collectTables, type SqliteDataMigrationClient } from "@executor-js/sdk"; +import { createSqliteTestFumaDb } from "@executor-js/sdk/testing"; + +import { createD1ExecutorDb } from "./d1"; + +const makeRecordingD1 = ( + client: SqliteDataMigrationClient, +): { + readonly db: D1Database; + readonly statements: string[]; + readonly failWhen: (predicate: ((sql: string) => boolean) | null) => void; +} => { + const statements: string[] = []; + let failurePredicate: ((sql: string) => boolean) | null = null; + 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"); + } + }; + const prepare = (sql: string) => { + const statement = (args: readonly unknown[]): Record => ({ + bind: (...values: readonly unknown[]) => statement([...args, ...values]), + all: async () => { + record(sql); + const result = await client.execute({ sql, args }); + return { success: true, meta: {}, results: result.rows }; + }, + run: async () => { + record(sql); + await client.execute({ sql, args }); + return { success: true, meta: {}, results: [] }; + }, + }); + return statement([]); + }; + + // oxlint-disable-next-line executor/no-double-cast -- test double: only the D1 methods used by schema preparation and migrations are implemented + const db = { + prepare, + withSession: () => ({ prepare }), + } as unknown as D1Database; + return { + db, + statements, + failWhen: (predicate) => { + failurePredicate = predicate; + }, + }; +}; + +const isRuntimeSchemaStatement = (sql: string): boolean => { + const normalized = sql.trim().toUpperCase(); + if (normalized.startsWith("CREATE UNIQUE INDEX IF NOT EXISTS")) return true; + if (normalized.startsWith("ALTER TABLE") && normalized.includes(" ADD COLUMN ")) return true; + if (!normalized.startsWith("CREATE TABLE IF NOT EXISTS")) return false; + return !normalized.includes("DATA_MIGRATION"); +}; + +describe("createD1ExecutorDb", () => { + it("does not repeat runtime schema DDL after the current schema was prepared", async () => { + const sqlite = await createSqliteTestFumaDb({ tables: collectTables() }); + const { db, statements } = makeRecordingD1(sqlite.client); + + const first = await createD1ExecutorDb(db, undefined); + await first.close(); + const firstSchemaStatements = statements.filter(isRuntimeSchemaStatement); + expect(firstSchemaStatements.length).toBeGreaterThan(0); + + const secondStart = statements.length; + const second = await createD1ExecutorDb(db, undefined); + await second.close(); + const secondStatements = statements.slice(secondStart); + + expect(secondStatements.filter(isRuntimeSchemaStatement)).toEqual([]); + expect(secondStatements.some((sql) => sql.includes("SELECT name FROM data_migration"))).toBe( + true, + ); + + await sqlite.close(); + }); + + it("runs the schema ensure again when the generated fingerprint is stale", async () => { + const sqlite = await createSqliteTestFumaDb({ tables: collectTables() }); + const { db, statements } = makeRecordingD1(sqlite.client); + + const first = await createD1ExecutorDb(db, undefined); + await first.close(); + await sqlite.client.execute( + `UPDATE private_executor_cloudflare_schema_fingerprint + SET fingerprint = 'stale' + WHERE id = 'runtime-schema'`, + ); + + const secondStart = statements.length; + const second = await createD1ExecutorDb(db, undefined); + await second.close(); + const secondStatements = statements.slice(secondStart); + + expect( + secondStatements.some((sql) => sql.startsWith('CREATE TABLE IF NOT EXISTS "integration"')), + ).toBe(true); + + await sqlite.close(); + }); + + it("converges safely when concurrent opens see the same stale fingerprint", async () => { + const sqlite = await createSqliteTestFumaDb({ tables: collectTables() }); + const { db } = makeRecordingD1(sqlite.client); + const prepared = await createD1ExecutorDb(db, undefined); + await prepared.close(); + await sqlite.client.execute( + `UPDATE private_executor_cloudflare_schema_fingerprint + SET fingerprint = 'stale' + WHERE id = 'runtime-schema'`, + ); + + const handles = await Promise.all([ + createD1ExecutorDb(db, undefined), + createD1ExecutorDb(db, undefined), + ]); + await Promise.all(handles.map((handle) => handle.close())); + + const fingerprints = await sqlite.client.execute( + `SELECT fingerprint + FROM private_executor_cloudflare_schema_fingerprint + WHERE id = 'runtime-schema'`, + ); + expect(fingerprints.rows).toHaveLength(1); + expect(fingerprints.rows[0]?.fingerprint).not.toBe("stale"); + + await sqlite.close(); + }); + + it("stamps only after compatibility migrations leave the final schema current", async () => { + const sqlite = await createSqliteTestFumaDb({ tables: collectTables() }); + await sqlite.client.execute("DROP TABLE connection"); + await sqlite.client.execute(` + CREATE TABLE connection ( + integration text NOT NULL, + name text NOT NULL, + template text NOT NULL, + provider text NOT NULL, + item_id text NOT NULL, + identity_label text, + description text, + tools_synced_at integer, + oauth_client text, + oauth_client_owner text, + refresh_item_id text, + expires_at integer, + oauth_scope text, + oauth_token_url text, + provider_state text, + created_at integer NOT NULL, + updated_at integer NOT NULL, + row_id text PRIMARY KEY NOT NULL, + tenant text NOT NULL, + owner text NOT NULL, + subject text NOT NULL + ) + `); + const { db } = makeRecordingD1(sqlite.client); + + const handle = await createD1ExecutorDb(db, undefined); + await handle.close(); + + const columns = await sqlite.client.execute("PRAGMA table_info('connection')"); + const columnNames = columns.rows.map((row) => row.name); + expect(columnNames).toContain("item_ids"); + expect(columnNames).toContain("last_health"); + expect(columnNames).not.toContain("item_id"); + const indexes = await sqlite.client.execute( + `SELECT name FROM sqlite_master + WHERE type = 'index' + AND name = 'connection_uidx'`, + ); + expect(indexes.rows).toEqual([{ name: "connection_uidx" }]); + + await sqlite.close(); + }); + + it("does not stamp a fingerprint when schema preparation fails", async () => { + const sqlite = await createSqliteTestFumaDb({ tables: collectTables() }); + const { db, failWhen } = makeRecordingD1(sqlite.client); + failWhen((sql) => sql.startsWith('CREATE TABLE IF NOT EXISTS "integration"')); + + await expect(createD1ExecutorDb(db, undefined)).rejects.toBeDefined(); + + const marker = await sqlite.client.execute( + `SELECT name FROM sqlite_master + WHERE type = 'table' + AND name = 'private_executor_cloudflare_schema_fingerprint'`, + ); + expect(marker.rows).toEqual([]); + + failWhen(null); + const recovered = await createD1ExecutorDb(db, undefined); + await recovered.close(); + + await sqlite.close(); + }); +}); diff --git a/apps/host-cloudflare/src/db/d1.ts b/apps/host-cloudflare/src/db/d1.ts index 0a2022626..43bc37587 100644 --- a/apps/host-cloudflare/src/db/d1.ts +++ b/apps/host-cloudflare/src/db/d1.ts @@ -1,6 +1,7 @@ import { drizzle } from "drizzle-orm/d1"; import { createDrizzleRuntimeSchemaFromTables, + createDrizzleRuntimeSchemaSqlFromTables, ensureDrizzleRuntimeSchemaFromTables, } from "@executor-js/fumadb/adapters/drizzle"; import type { D1Database, R2Bucket } from "@cloudflare/workers-types"; @@ -13,7 +14,64 @@ import { import { makeR2BlobStore } from "@executor-js/cloudflare/blob-store"; import { CLOUDFLARE_NAMESPACE, CLOUDFLARE_SCHEMA_VERSION } from "../config"; -import { runCloudflareDataMigrations } from "./data-migrations"; +import { prepareCloudflareD1Data } from "./data-migrations"; + +const SCHEMA_FINGERPRINT_TABLE = `private_${CLOUDFLARE_NAMESPACE}_schema_fingerprint`; +const SCHEMA_FINGERPRINT_ID = "runtime-schema"; + +const schemaFingerprint = async (statements: readonly string[]): Promise => { + const bytes = new TextEncoder().encode(statements.join("\0")); + const digest = await crypto.subtle.digest("SHA-256", bytes); + return Array.from(new Uint8Array(digest), (byte) => byte.toString(16).padStart(2, "0")).join(""); +}; + +const isMissingFingerprintTable = (error: unknown): boolean => { + for (let current = error, depth = 0; current != null && depth < 8; depth += 1) { + const record = + typeof current === "object" ? (current as { message?: unknown; cause?: unknown }) : null; + const message = typeof record?.message === "string" ? record.message : String(current); + if (/no such table/i.test(message) && message.includes(SCHEMA_FINGERPRINT_TABLE)) return true; + current = record?.cause ?? null; + } + return false; +}; + +const readPreparedSchemaFingerprint = async (db: D1Database): Promise => { + const session = db.withSession("first-primary"); + // oxlint-disable-next-line executor/no-try-catch-or-throw -- boundary: a missing marker table is the expected first-boot signal; every other D1 failure must still reject startup + try { + const result = await session + .prepare(`SELECT fingerprint FROM "${SCHEMA_FINGERPRINT_TABLE}" WHERE id = ?`) + .bind(SCHEMA_FINGERPRINT_ID) + .all<{ readonly fingerprint?: unknown }>(); + const value = result.results[0]?.fingerprint; + return typeof value === "string" ? value : null; + } catch (error) { + if (isMissingFingerprintTable(error)) return null; + // oxlint-disable-next-line executor/no-try-catch-or-throw -- boundary: preserve the original D1 adapter rejection for every failure except the expected absent first-boot marker + throw error; + } +}; + +const storePreparedSchemaFingerprint = async ( + db: D1Database, + fingerprint: string, +): Promise => { + const session = db.withSession("first-primary"); + await session + .prepare( + `CREATE TABLE IF NOT EXISTS "${SCHEMA_FINGERPRINT_TABLE}" (id text PRIMARY KEY NOT NULL, fingerprint text NOT NULL)`, + ) + .run(); + await session + .prepare( + `INSERT INTO "${SCHEMA_FINGERPRINT_TABLE}" (id, fingerprint) + VALUES (?, ?) + ON CONFLICT(id) DO UPDATE SET fingerprint = excluded.fingerprint`, + ) + .bind(SCHEMA_FINGERPRINT_ID, fingerprint) + .run(); +}; // --------------------------------------------------------------------------- // D1 DbProvider handle — the CF-native swap for self-host's libSQL handle. @@ -42,10 +100,31 @@ export const createD1ExecutorDb = async ( // 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 - // exposes one. The bring-up is idempotent `CREATE TABLE IF NOT EXISTS`, so run - // it WITHOUT a transaction by handing the ensure a run-only view of the handle. - await ensureDrizzleRuntimeSchemaFromTables({ run: (query) => drizzleDb.run(query) }, options); - await runCloudflareDataMigrations(db, blobs); + // exposes one. A generated fingerprint lets every Worker/DO isolate prove the + // expected generated schema was already prepared with one primary read instead of + // replaying dozens of idempotent CREATE/ALTER statements on every database + // open. The marker is stored only after the ensure, data migrations, and any + // required post-compatibility ensure succeed. Data migrations retain their + // own ledger and still run below on every open. + const expectedFingerprint = await schemaFingerprint( + createDrizzleRuntimeSchemaSqlFromTables(options), + ); + const preparedFingerprint = await readPreparedSchemaFingerprint(db); + const requiresSchemaPreparation = preparedFingerprint !== expectedFingerprint; + if (requiresSchemaPreparation) { + await ensureDrizzleRuntimeSchemaFromTables({ run: (query) => drizzleDb.run(query) }, options); + } + const migrationResult = await prepareCloudflareD1Data(db, blobs); + if (requiresSchemaPreparation) { + // Compatibility migrations can rebuild a legacy table. Re-run the + // idempotent ensure before stamping so the database matches the generated + // schema after migration, including newer nullable columns the legacy + // rebuild did not know about. + if (migrationResult.schemaChanged) { + await ensureDrizzleRuntimeSchemaFromTables({ run: (query) => drizzleDb.run(query) }, options); + } + await storePreparedSchemaFingerprint(db, expectedFingerprint); + } // `interactiveTransactions: false` — D1 rejects interactive transactions, so // the fuma adapter runs transaction callbacks directly (auto-commit per diff --git a/apps/host-cloudflare/src/db/data-migrations.ts b/apps/host-cloudflare/src/db/data-migrations.ts index 85f48b371..d7c63b286 100644 --- a/apps/host-cloudflare/src/db/data-migrations.ts +++ b/apps/host-cloudflare/src/db/data-migrations.ts @@ -108,7 +108,8 @@ const rebuildLegacyConnectionTable = async (db: D1Database): Promise => { } }; -export const ensureCloudflareD1SchemaCompatibility = async (db: D1Database): Promise => { +export const ensureCloudflareD1SchemaCompatibility = async (db: D1Database): Promise => { + let schemaChanged = false; const integrationColumns = await tableColumns(db, "integration"); if (integrationColumns.has("config")) { await db @@ -122,14 +123,17 @@ export const ensureCloudflareD1SchemaCompatibility = async (db: D1Database): Pro } const connectionColumns = await tableColumns(db, "connection"); - if (connectionColumns.size === 0) return; + if (connectionColumns.size === 0) return schemaChanged; if (!connectionColumns.has("item_ids")) { await db.prepare(`ALTER TABLE connection ADD COLUMN item_ids json NOT NULL DEFAULT '{}'`).run(); + schemaChanged = true; } const updatedConnectionColumns = await tableColumns(db, "connection"); if (updatedConnectionColumns.has("item_id")) { await rebuildLegacyConnectionTable(db); + schemaChanged = true; } + return schemaChanged; }; const r2ObjectName = (tenant: string, pluginId: string, key: string): string => @@ -248,14 +252,24 @@ const cloudflareDataMigrations = (bucket: R2Bucket | undefined): readonly Sqlite openApiNdjsonOutputDataMigration, ]; -export const runCloudflareDataMigrations = ( +export const prepareCloudflareD1Data = ( db: D1Database, bucket: R2Bucket | undefined, -): Promise => +): Promise<{ + readonly appliedMigrations: readonly string[]; + readonly schemaChanged: boolean; +}> => Effect.runPromise( Effect.promise(() => ensureCloudflareD1SchemaCompatibility(db)).pipe( - Effect.flatMap(() => - runSqliteDataMigrations(d1DataMigrationClient(db), cloudflareDataMigrations(bucket)), + Effect.flatMap((schemaChanged) => + runSqliteDataMigrations(d1DataMigrationClient(db), cloudflareDataMigrations(bucket)).pipe( + Effect.map((appliedMigrations) => ({ appliedMigrations, schemaChanged })), + ), ), ), ); + +export const runCloudflareDataMigrations = async ( + db: D1Database, + bucket: R2Bucket | undefined, +): Promise => (await prepareCloudflareD1Data(db, bucket)).appliedMigrations;