Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 0 additions & 6 deletions apps/host-cloudflare/src/app.ts
Original file line number Diff line number Diff line change
Expand Up @@ -26,7 +26,6 @@ import { ErrorCaptureLive } from "./observability";
import { cloudflareAccountMiddleware } from "./account/account-provider";
import { makeCloudflareApprovalHandler } from "./mcp";
import { makeCloudflareMcpAgentHandler } from "./mcp/agent-handler";
import { preloadQuickJs } from "./quickjs";

// ===========================================================================
// The Cloudflare host, as ONE `ExecutorApp.make` call: the 4th app alongside
Expand Down Expand Up @@ -67,7 +66,6 @@ export const reconcileInternalIntegrationsOnce = (env: CloudflareEnv): Promise<v
const config = loadConfig(env);
const accountId = internalPrincipal(config, { subject: null, email: null })?.accountId;
if (config.internalIntegrations.length === 0 || accountId === undefined) return;
yield* Effect.promise(() => preloadQuickJs());
const handle = yield* Effect.promise(() => openDb(env));
const executor = yield* makeScopedExecutor<CloudflarePlugins>(
accountId,
Expand All @@ -94,10 +92,6 @@ export const makeCloudflareApp = async (
) => {
const plugins = makeCloudflarePlugins(config.secretKey);

// Load the Workers-compatible (WASM-inlined) QuickJS variant before any
// executor is built, the default variant cannot fetch its .wasm on Workers.
await preloadQuickJs();

// Open and idempotently bring up the D1 schema once. This is the long-lived
// handle the per-request scoped executor reads through the DbProvider seam.
const handle = await openDb(env);
Expand Down
165 changes: 125 additions & 40 deletions apps/host-cloudflare/src/db/d1.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,14 @@ import { collectTables, type SqliteDataMigrationClient } from "@executor-js/sdk"
import { createSqliteTestFumaDb } from "@executor-js/sdk/testing";

import { createD1ExecutorDb } from "./d1";
import { cloudflareDataMigrationIds } from "./data-migrations";

const readMarker = async (client: SqliteDataMigrationClient): Promise<string> => {
const result = await client.execute(
`SELECT fingerprint FROM private_executor_cloudflare_schema_fingerprint WHERE id = 'runtime-schema'`,
);
return String(result.rows[0]?.fingerprint);
};

const makeRecordingD1 = (
client: SqliteDataMigrationClient,
Expand Down Expand Up @@ -63,46 +71,99 @@ const isRuntimeSchemaStatement = (sql: string): boolean => {
return !normalized.includes("DATA_MIGRATION");
};

const ranDataMigrations = (statements: readonly string[]): boolean =>
statements.some((sql) => sql.includes("SELECT name FROM data_migration"));

const setMarker = (client: SqliteDataMigrationClient, fingerprint: string) =>
client.execute({
sql: `UPDATE private_executor_cloudflare_schema_fingerprint SET fingerprint = ? WHERE id = 'runtime-schema'`,
args: [fingerprint],
});

// Opens the database once so it is current, then opens it again and returns
// only what the second open sent to D1.
const openTwice = async (
db: D1Database,
statements: string[],
between: () => Promise<unknown> = async () => {},
): Promise<string[]> => {
const first = await createD1ExecutorDb(db, undefined);
await first.close();
await between();
const start = statements.length;
const second = await createD1ExecutorDb(db, undefined);
await second.close();
return statements.slice(start);
};

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);
const secondStatements = await openTwice(db, statements);

expect(statements.filter(isRuntimeSchemaStatement).length).toBeGreaterThan(0);
expect(secondStatements.filter(isRuntimeSchemaStatement)).toEqual([]);
expect(secondStatements.some((sql) => sql.includes("SELECT name FROM data_migration"))).toBe(
true,

await sqlite.close();
});

it("opens a current database with exactly one D1 query", async () => {
const sqlite = await createSqliteTestFumaDb({ tables: collectTables() });
const { db, statements } = makeRecordingD1(sqlite.client);

const secondStatements = await openTwice(db, statements);

expect(secondStatements).toHaveLength(1);
expect(secondStatements[0]).toContain("SELECT fingerprint");

await sqlite.close();
});

it("records every known data migration in the ledger", async () => {
const sqlite = await createSqliteTestFumaDb({ tables: collectTables() });
const { db } = makeRecordingD1(sqlite.client);

const handle = await createD1ExecutorDb(db, undefined);
await handle.close();

const ledger = await sqlite.client.execute(`SELECT name FROM data_migration`);
expect(ledger.rows.map((row) => row.name)).toEqual(
expect.arrayContaining([...cloudflareDataMigrationIds]),
);

await sqlite.close();
});

it("runs the data migrations again when the stored fingerprint predates the migration list", async () => {
const sqlite = await createSqliteTestFumaDb({ tables: collectTables() });
const { db, statements } = makeRecordingD1(sqlite.client);
const missing = cloudflareDataMigrationIds[cloudflareDataMigrationIds.length - 1];

const secondStatements = await openTwice(db, statements, async () => {
await sqlite.client.execute({
sql: "DELETE FROM data_migration WHERE name = ?",
args: [missing],
});
await setMarker(sqlite.client, "older");
});

expect(ranDataMigrations(secondStatements)).toBe(true);
const ledger = await sqlite.client.execute(`SELECT name FROM data_migration`);
expect(ledger.rows.map((row) => row.name)).toContain(missing);

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 secondStatements = await openTwice(db, statements, () =>
setMarker(sqlite.client, "stale"),
);

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);
Expand All @@ -112,20 +173,14 @@ describe("createD1ExecutorDb", () => {

it("adds the compiled artifact columns to a database that predates them", async () => {
const sqlite = await createSqliteTestFumaDb({ tables: collectTables() });
const { db } = makeRecordingD1(sqlite.client);
const first = await createD1ExecutorDb(db, undefined);
await first.close();
for (const column of ["compiled_fingerprint", "compiled_js", "compiled_css"]) {
await sqlite.client.execute(`ALTER TABLE "artifact" DROP COLUMN "${column}"`);
}
await sqlite.client.execute(
`UPDATE private_executor_cloudflare_schema_fingerprint
SET fingerprint = 'stale'
WHERE id = 'runtime-schema'`,
);
const { db, statements } = makeRecordingD1(sqlite.client);

const second = await createD1ExecutorDb(db, undefined);
await second.close();
await openTwice(db, statements, async () => {
for (const column of ["compiled_fingerprint", "compiled_js", "compiled_css"]) {
await sqlite.client.execute(`ALTER TABLE "artifact" DROP COLUMN "${column}"`);
}
await setMarker(sqlite.client, "stale");
});

const columns = await sqlite.client.execute(`PRAGMA table_info("artifact")`);
const names = columns.rows.map((row) => String(row.name));
Expand All @@ -141,11 +196,7 @@ describe("createD1ExecutorDb", () => {
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'`,
);
await setMarker(sqlite.client, "stale");

const handles = await Promise.all([
createD1ExecutorDb(db, undefined),
Expand Down Expand Up @@ -256,4 +307,38 @@ describe("createD1ExecutorDb", () => {

await sqlite.close();
});

it("rewrites the marker after preparing a database whose marker is out of date", async () => {
const sqlite = await createSqliteTestFumaDb({ tables: collectTables() });
const { db, statements } = makeRecordingD1(sqlite.client);
let currentMarker = "";

const secondStatements = await openTwice(db, statements, async () => {
currentMarker = await readMarker(sqlite.client);
await setMarker(sqlite.client, "older");
});

expect(ranDataMigrations(secondStatements)).toBe(true);
expect(await readMarker(sqlite.client)).toBe(currentMarker);

await sqlite.close();
});

it("keeps the old marker when data preparation fails so the next open retries", async () => {
const sqlite = await createSqliteTestFumaDb({ tables: collectTables() });
const { db, statements, failWhen } = makeRecordingD1(sqlite.client);
let currentMarker = "";
await openTwice(db, statements, async () => {
currentMarker = await readMarker(sqlite.client);
await setMarker(sqlite.client, "older");
failWhen((sql) => sql.includes("SELECT name FROM data_migration"));
await expect(createD1ExecutorDb(db, undefined)).rejects.toBeDefined();
expect(await readMarker(sqlite.client)).toBe("older");
failWhen(null);
});

expect(await readMarker(sqlite.client)).toBe(currentMarker);

await sqlite.close();
});
});
42 changes: 31 additions & 11 deletions apps/host-cloudflare/src/db/d1.ts
Original file line number Diff line number Diff line change
Expand Up @@ -14,7 +14,11 @@ import {
import { makeR2BlobStore } from "@executor-js/cloudflare/blob-store";

import { CLOUDFLARE_NAMESPACE, CLOUDFLARE_SCHEMA_VERSION } from "../config";
import { prepareCloudflareD1Data } from "./data-migrations";
import {
cloudflareDataMigrationIds,
ensureCloudflareD1SchemaCompatibility,
prepareCloudflareD1Data,
} from "./data-migrations";
import { withD1TransientRetry } from "./d1-retry";

const SCHEMA_FINGERPRINT_TABLE = `private_${CLOUDFLARE_NAMESPACE}_schema_fingerprint`;
Expand All @@ -26,6 +30,22 @@ const schemaFingerprint = async (statements: readonly string[]): Promise<string>
return Array.from(new Uint8Array(digest), (byte) => byte.toString(16).padStart(2, "0")).join("");
};

// Every open in an isolate uses the same fixed options, so the hash is
// computed once. Hashing the compatibility function's source makes a change to
// the repair code change the fingerprint without a hand-bumped version.
let expectedFingerprintCache: Promise<string> | null = null;

const expectedSchemaFingerprint = (
options: Parameters<typeof createDrizzleRuntimeSchemaSqlFromTables>[0],
): Promise<string> => {
expectedFingerprintCache ??= schemaFingerprint([
...createDrizzleRuntimeSchemaSqlFromTables(options),
`compatibility:${ensureCloudflareD1SchemaCompatibility.toString()}`,
`data-migrations:${cloudflareDataMigrationIds.join(",")}`,
]);
return expectedFingerprintCache;
};

const isMissingFingerprintTable = (error: unknown): boolean => {
for (let current = error, depth = 0; current != null && depth < 8; depth += 1) {
const record =
Expand Down Expand Up @@ -105,19 +125,19 @@ export const createD1ExecutorDb = async (
// 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);
// open. The fingerprint also covers the compatibility-check source and the
// ordered data-migration ids, so a current database opens with that one read
// and skips both. The marker is stored only after the ensure, compatibility
// check, data migrations, and any required post-compatibility ensure succeed.
// Data migrations keep their own ledger for databases that need them.
const [expectedFingerprint, preparedFingerprint] = await Promise.all([
expectedSchemaFingerprint(options),
readPreparedSchemaFingerprint(db),
]);
const requiresSchemaPreparation = preparedFingerprint !== expectedFingerprint;
if (requiresSchemaPreparation) {
await ensureDrizzleRuntimeSchemaFromTables({ run: (query) => drizzleDb.run(query) }, options);
}
const migrationResult = await prepareCloudflareD1Data(db, blobs);
if (requiresSchemaPreparation) {
const migrationResult = await prepareCloudflareD1Data(db, blobs);
// 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
Expand Down
10 changes: 10 additions & 0 deletions apps/host-cloudflare/src/db/data-migrations.ts
Original file line number Diff line number Diff line change
Expand Up @@ -256,6 +256,16 @@ const cloudflareDataMigrations = (bucket: R2Bucket | undefined): readonly Sqlite
encryptedSecretsRepartitionDataMigration,
];

// Ordered names of every data migration this build knows. The ledger already
// makes a migration run once, and the bucket only decides how it copies blobs,
// so the names alone describe what a fully prepared database looks like.
export const cloudflareDataMigrationIds: readonly string[] = [
googleOpenApiOwnershipDataMigration.name,
providerServiceSplitDataMigration.name,
openApiNdjsonOutputDataMigration.name,
encryptedSecretsRepartitionDataMigration.name,
];

export const prepareCloudflareD1Data = (
db: D1Database,
bucket: R2Bucket | undefined,
Expand Down
28 changes: 28 additions & 0 deletions apps/host-cloudflare/src/execution.test.ts
Original file line number Diff line number Diff line change
@@ -0,0 +1,28 @@
import { describe, expect, it } from "@effect/vitest";
// oxlint-disable-next-line executor/no-vitest-import -- vi.mock hoisting needs the vitest import
import { vi } from "vitest";
import { Effect } from "effect";

import { CodeExecutorProvider } from "@executor-js/api/server";

const { loadQuickJs } = vi.hoisted(() => ({ loadQuickJs: vi.fn() }));

vi.mock("./quickjs", () => ({ loadQuickJs }));
vi.mock("./mcp", () => ({ McpExecutionOwnerDirectoryDO: class {}, McpSessionDO: class {} }));
vi.mock("./app", () => ({
makeCloudflareApp: vi.fn(),
reconcileInternalIntegrationsOnce: vi.fn(),
}));

await import("./worker");
const { CloudflareCodeExecutorProvider } = await import("./execution");

describe("CloudflareCodeExecutorProvider", () => {
it.effect("does not load QuickJS when the worker starts or the provider is built", () =>
Effect.gen(function* () {
yield* CodeExecutorProvider.asEffect().pipe(Effect.provide(CloudflareCodeExecutorProvider));

expect(loadQuickJs).not.toHaveBeenCalled();
}),
);
});
4 changes: 0 additions & 4 deletions apps/host-cloudflare/src/mcp/session-durable-object.ts
Original file line number Diff line number Diff line change
Expand Up @@ -30,7 +30,6 @@ import type { ResumeResponse } from "@executor-js/execution";
import { loadConfig, type CloudflareConfig, type CloudflareEnv } from "../config";
import { createD1ExecutorDb } from "../db/d1";
import { makeCloudflareExecutionStackLayer, makeExecutionStack } from "../execution";
import { preloadQuickJs } from "../quickjs";

// ---------------------------------------------------------------------------
// Cloudflare (self-host) MCP Session Durable Object — the host-cloudflare
Expand Down Expand Up @@ -146,9 +145,6 @@ export class McpSessionDO extends McpAgentSessionDOBase<CloudflareEnv, CfSession
const config = this.cfConfig;
const self = this;
return Effect.gen(function* () {
// QuickJS-WASM must be loaded before the executor layer builds it (the
// default variant can't fetch its .wasm on Workers). Idempotent per isolate.
yield* Effect.promise(() => preloadQuickJs());
const { engine, executor } = yield* makeExecutionStack(
sessionMeta.userId,
sessionMeta.organizationId,
Expand Down
Loading
Loading