From b257c9ece5635a5c70786944a49d2ffeb9990a14 Mon Sep 17 00:00:00 2001 From: fforres Date: Sat, 17 Oct 2026 17:38:00 -0700 Subject: [PATCH 1/2] perf(host-cloudflare): reconcile internal integrations on a cron instead of at isolate boot Cold isolates waited about 4 s on a sequential reconcile of every internal integration before serving any request (API, MCP, internal door). Reconcile now runs every minute from a cron trigger via a scheduled handler, all entries concurrently, and the fetch path no longer touches it. A new internal integration or a changed tool list now appears within about a minute instead of at isolate boot. Claude-Session: https://claude.ai/code/session_019ZE3j9YzRKxT4LaiDCBv6S --- apps/host-cloudflare/src/app.ts | 60 +++++++++--------- .../internal-integrations/reconcile.test.ts | 39 +++++++++++- .../src/internal-integrations/reconcile.ts | 14 +++-- apps/host-cloudflare/src/worker.test.ts | 62 +++++++++++++++++++ apps/host-cloudflare/src/worker.ts | 9 ++- apps/host-cloudflare/wrangler.jsonc | 1 + 6 files changed, 147 insertions(+), 38 deletions(-) create mode 100644 apps/host-cloudflare/src/worker.test.ts diff --git a/apps/host-cloudflare/src/app.ts b/apps/host-cloudflare/src/app.ts index d15860fa33..5504223387 100644 --- a/apps/host-cloudflare/src/app.ts +++ b/apps/host-cloudflare/src/app.ts @@ -4,7 +4,6 @@ import { HttpEffect, HttpRouter } from "effect/unstable/http"; import { dbProviderLayer, ExecutorApp, - type ExecutorDbHandle, makeScopedExecutor, textFailureStrategy, } from "@executor-js/api/server"; @@ -42,34 +41,13 @@ import { preloadQuickJs } from "./quickjs"; // so the providers close over it instead of reading process.env. // =========================================================================== -let internalIntegrationsReconciled: Promise | null = null; - -// Once per isolate, by a system executor that may write org rows. Also refreshes -// existing declared catalogs whose tool list changed. Never fails the boot. -const reconcileOncePerIsolate = ( - config: CloudflareConfig, - dbHandle: ExecutorDbHandle, - env: CloudflareEnv, -): Promise => { - if (config.internalIntegrations.length === 0) return Promise.resolve(); - internalIntegrationsReconciled ??= Effect.runPromise( - Effect.gen(function* () { - const accountId = internalPrincipal(config, { subject: null, email: null })?.accountId; - if (accountId === undefined) return; - const executor = yield* makeScopedExecutor( - accountId, - config.organizationId, - config.organizationName, - { orgWrites: "allowed" }, - ); - yield* reconcileInternalIntegrations(executor, config.internalIntegrations, { - listTools: makeInternalListTools(config.internalHosts), - store: env.BLOBS, - }).pipe(Effect.ensuring(executor.close().pipe(Effect.ignore))); - }).pipe( - Effect.provide(dbProviderLayer(Effect.succeed(dbHandle))), - Effect.provide(makeCloudflarePluginsProvider(config)), - Effect.provide(makeCloudflareHostConfig(config)), +// One reconcile pass, run from the cron trigger (never on the request path), by a +// system executor that may write org rows. Also refreshes existing declared +// catalogs whose tool list changed. Never throws. +export const reconcileInternalIntegrationsOnce = (env: CloudflareEnv): Promise => + Effect.runPromise( + Effect.sync(() => loadConfig(env)).pipe( + Effect.flatMap(reconcileWithConfig(env)), Effect.catchCause((cause) => Effect.logWarning("internal integration reconcile aborted", cause).pipe( Effect.annotateLogs({ step: "reconcile-internal-integrations" }), @@ -77,7 +55,28 @@ const reconcileOncePerIsolate = ( ), ), ); - return internalIntegrationsReconciled; + +const reconcileWithConfig = (env: CloudflareEnv) => (config: CloudflareConfig) => { + if (config.internalIntegrations.length === 0) return Effect.void; + return Effect.gen(function* () { + const accountId = internalPrincipal(config, { subject: null, email: null })?.accountId; + if (accountId === undefined) return; + yield* Effect.promise(() => preloadQuickJs()); + const dbHandle = yield* Effect.promise(() => createD1ExecutorDb(env.DB, env.BLOBS)); + const executor = yield* makeScopedExecutor( + accountId, + config.organizationId, + config.organizationName, + { orgWrites: "allowed" }, + ).pipe(Effect.provide(dbProviderLayer(Effect.succeed(dbHandle)))); + yield* reconcileInternalIntegrations(executor, config.internalIntegrations, { + listTools: makeInternalListTools(config.internalHosts), + store: env.BLOBS, + }).pipe(Effect.ensuring(executor.close().pipe(Effect.ignore))); + }).pipe( + Effect.provide(makeCloudflarePluginsProvider(config)), + Effect.provide(makeCloudflareHostConfig(config)), + ); }; export const makeCloudflareApp = async ( @@ -93,7 +92,6 @@ export const makeCloudflareApp = async ( // 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 dbHandle = await createD1ExecutorDb(env.DB, env.BLOBS); - await reconcileOncePerIsolate(config, dbHandle, env); const identityLayer = cloudflareAccessIdentityLayer(config); const mcpAgentHandler = makeCloudflareMcpAgentHandler(config); const approvalHandler = makeCloudflareApprovalHandler(config, env); diff --git a/apps/host-cloudflare/src/internal-integrations/reconcile.test.ts b/apps/host-cloudflare/src/internal-integrations/reconcile.test.ts index 7af283b651..749fad1a5a 100644 --- a/apps/host-cloudflare/src/internal-integrations/reconcile.test.ts +++ b/apps/host-cloudflare/src/internal-integrations/reconcile.test.ts @@ -1,5 +1,5 @@ import { describe, expect, it } from "@effect/vitest"; -import { Effect, Option, Schema } from "effect"; +import { Deferred, Effect, Option, Schema } from "effect"; import { WebStandardStreamableHTTPServerTransport } from "@modelcontextprotocol/sdk/server/webStandardStreamableHttp.js"; import { createExecutor, IntegrationSlug } from "@executor-js/sdk"; @@ -270,6 +270,43 @@ describe("reconcileInternalIntegrations", () => { ); }); +describe("reconcileInternalIntegrations concurrency", () => { + it.live("starts every entry before any of them is released", () => + Effect.scoped( + Effect.gen(function* () { + const binding = makeBinding(); + const { executor } = yield* makeExecutor(binding); + const entries = ["alpha", "beta", "gamma"].map( + (slug): InternalIntegration => ({ + ...FIXTURE, + slug, + name: slug, + endpoint: `https://tools.internal/mcp/${slug}`, + }), + ); + const gate = yield* Deferred.make(); + const started: string[] = []; + const listTools = (endpoint: string) => + Effect.gen(function* () { + started.push(endpoint); + if (started.length === entries.length) yield* Deferred.succeed(gate, undefined); + yield* Deferred.await(gate); + return [] as ReadonlyArray; + }); + + const outcome = yield* reconcileInternalIntegrations(executor, entries, { listTools }); + + expect(started.sort()).toEqual([ + "https://tools.internal/mcp/alpha", + "https://tools.internal/mcp/beta", + "https://tools.internal/mcp/gamma", + ]); + expect(outcome.map(([slug]) => slug)).toEqual(["alpha", "beta", "gamma"]); + }), + ), + ); +}); + describe("content-hash catalog refresh", () => { const created = (binding: ReturnType, store: HashStore) => Effect.gen(function* () { diff --git a/apps/host-cloudflare/src/internal-integrations/reconcile.ts b/apps/host-cloudflare/src/internal-integrations/reconcile.ts index ef013535a3..bb90599770 100644 --- a/apps/host-cloudflare/src/internal-integrations/reconcile.ts +++ b/apps/host-cloudflare/src/internal-integrations/reconcile.ts @@ -237,15 +237,19 @@ const reconcileOne = ( * stored endpoint equals the declared one, lists the current tools and * refreshes the `main` connection only when their content hash differs from the * stored one; the hash is stored after a successful refresh. Never updates or - * deletes anything, and never fails the caller. + * deletes anything, and never fails the caller. Entries run concurrently; each + * catches its own failure. */ export const reconcileInternalIntegrations = ( executor: InternalIntegrationExecutor, entries: readonly InternalIntegration[], deps: ReconcileDeps = {}, ): Effect.Effect> => - Effect.forEach(entries, (entry) => - reconcileOne(executor, entry, deps).pipe( - Effect.map((outcome) => [entry.slug, outcome] as const), - ), + Effect.forEach( + entries, + (entry) => + reconcileOne(executor, entry, deps).pipe( + Effect.map((outcome) => [entry.slug, outcome] as const), + ), + { concurrency: "unbounded" }, ); diff --git a/apps/host-cloudflare/src/worker.test.ts b/apps/host-cloudflare/src/worker.test.ts new file mode 100644 index 0000000000..c664109317 --- /dev/null +++ b/apps/host-cloudflare/src/worker.test.ts @@ -0,0 +1,62 @@ +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 type { CloudflareEnv } from "./config"; + +const { makeCloudflareApp, reconcileInternalIntegrationsOnce } = vi.hoisted(() => ({ + makeCloudflareApp: vi.fn(() => + Promise.resolve({ + toWebHandler: () => ({ handler: () => Promise.resolve(new Response("ok")) }), + mcpAgentHandler: () => Promise.resolve(new Response("mcp")), + }), + ), + reconcileInternalIntegrationsOnce: vi.fn(() => Promise.resolve()), +})); + +vi.mock("./mcp", () => ({ McpExecutionOwnerDirectoryDO: class {}, McpSessionDO: class {} })); +vi.mock("./app", () => ({ makeCloudflareApp, reconcileInternalIntegrationsOnce })); + +const { default: worker } = await import("./worker"); + +const makeEnv = (vars: Partial): CloudflareEnv => + // oxlint-disable-next-line executor/no-double-cast -- test double: only the vars the code under test reads are set + vars as unknown as CloudflareEnv; + +const ctx = (waited: Promise[]): ExecutionContext => { + const context = { waitUntil: (promise: Promise) => waited.push(promise) }; + // oxlint-disable-next-line executor/no-double-cast -- test double: only waitUntil is used + return context as unknown as ExecutionContext; +}; + +describe("worker", () => { + it("serves a fetch without running a reconcile pass", async () => { + const env = makeEnv({ + ENABLE_DEV_AUTH: "true", + EXECUTOR_SECRET_KEY: "test-secret-key-0123456789abcdef", + }); + + const response = await worker.fetch( + new Request("https://executor.test/api/scopes"), + env, + ctx([]), + ); + + expect(await response.text()).toBe("ok"); + expect(makeCloudflareApp).toHaveBeenCalledTimes(1); + expect(reconcileInternalIntegrationsOnce).not.toHaveBeenCalled(); + }); + + it("scheduled runs one reconcile pass inside waitUntil", async () => { + const env = makeEnv({}); + const waited: Promise[] = []; + + // oxlint-disable-next-line executor/no-double-cast -- test double: the handler ignores the controller + worker.scheduled({} as unknown as ScheduledController, env, ctx(waited)); + + expect(reconcileInternalIntegrationsOnce).toHaveBeenCalledTimes(1); + expect(reconcileInternalIntegrationsOnce).toHaveBeenCalledWith(env); + expect(waited).toHaveLength(1); + await Promise.all(waited); + }); +}); diff --git a/apps/host-cloudflare/src/worker.ts b/apps/host-cloudflare/src/worker.ts index fd0b402f0e..6ed8951936 100644 --- a/apps/host-cloudflare/src/worker.ts +++ b/apps/host-cloudflare/src/worker.ts @@ -1,6 +1,6 @@ import { WorkerEntrypoint } from "cloudflare:workers"; -import { makeCloudflareApp } from "./app"; +import { makeCloudflareApp, reconcileInternalIntegrationsOnce } from "./app"; import { cloudflareAccessConfigErrorMessage, loadConfig, @@ -94,4 +94,11 @@ export class ExecutorInternal extends WorkerEntrypoint { export default { fetch: (request: Request, env: CloudflareEnv, ctx: ExecutionContext): Promise => serveRequest(request, env, ctx, resolvePublic), + scheduled: ( + _controller: ScheduledController, + env: CloudflareEnv, + ctx: ExecutionContext, + ): void => { + ctx.waitUntil(reconcileInternalIntegrationsOnce(env)); + }, }; diff --git a/apps/host-cloudflare/wrangler.jsonc b/apps/host-cloudflare/wrangler.jsonc index cb3838fa86..d16228ed6d 100644 --- a/apps/host-cloudflare/wrangler.jsonc +++ b/apps/host-cloudflare/wrangler.jsonc @@ -7,6 +7,7 @@ // Access configuration is set per installation after the first deploy. // Preserve those live bindings when this source config does not declare them. "keep_vars": true, + "triggers": { "crons": ["* * * * *"] }, "observability": { "enabled": true, "traces": { "enabled": true, "head_sampling_rate": 1 }, From 5a0e117bdb61a0b389b78205be45334f0aa15cb6 Mon Sep 17 00:00:00 2001 From: fforres Date: Sat, 17 Oct 2026 17:58:15 -0700 Subject: [PATCH 2/2] refactor(host-cloudflare): share the D1 handle and execution stack with the reconcile cron The cron reuses the per-isolate D1 handle instead of reopening D1 every minute (a failed open is retried), builds its executor on makeCloudflareExecutionStackLayer, and loads config inside a single Effect.gen so a throw still ends in the "never throws" warning. Claude-Session: https://claude.ai/code/session_019ZE3j9YzRKxT4LaiDCBv6S --- apps/host-cloudflare/src/app.ts | 63 +++++++++++++++++++-------------- 1 file changed, 36 insertions(+), 27 deletions(-) diff --git a/apps/host-cloudflare/src/app.ts b/apps/host-cloudflare/src/app.ts index 5504223387..af62a386a5 100644 --- a/apps/host-cloudflare/src/app.ts +++ b/apps/host-cloudflare/src/app.ts @@ -6,6 +6,7 @@ import { ExecutorApp, makeScopedExecutor, textFailureStrategy, + type ExecutorDbHandle, } from "@executor-js/api/server"; import { internalPrincipal } from "./auth/cloudflare-access"; @@ -17,6 +18,7 @@ import { createD1ExecutorDb } from "./db/d1"; import { cloudflareAccessIdentityLayer } from "./auth/cloudflare-access"; import { CloudflareCodeExecutorProvider, + makeCloudflareExecutionStackLayer, makeCloudflareHostConfig, makeCloudflarePluginsProvider, } from "./execution"; @@ -41,13 +43,43 @@ import { preloadQuickJs } from "./quickjs"; // so the providers close over it instead of reading process.env. // =========================================================================== +// One D1 handle per isolate, shared by the request path and the cron. A failed +// open is dropped so the next caller retries. +let dbHandle: Promise | null = null; + +const openDb = (env: CloudflareEnv): Promise => { + if (dbHandle === null) { + const opening = createD1ExecutorDb(env.DB, env.BLOBS); + dbHandle = opening; + void opening.then(undefined, () => { + if (dbHandle === opening) dbHandle = null; + }); + } + return dbHandle; +}; + // One reconcile pass, run from the cron trigger (never on the request path), by a // system executor that may write org rows. Also refreshes existing declared // catalogs whose tool list changed. Never throws. export const reconcileInternalIntegrationsOnce = (env: CloudflareEnv): Promise => Effect.runPromise( - Effect.sync(() => loadConfig(env)).pipe( - Effect.flatMap(reconcileWithConfig(env)), + Effect.gen(function* () { + 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( + accountId, + config.organizationId, + config.organizationName, + { orgWrites: "allowed" }, + ).pipe(Effect.provide(makeCloudflareExecutionStackLayer(config, handle))); + yield* reconcileInternalIntegrations(executor, config.internalIntegrations, { + listTools: makeInternalListTools(config.internalHosts), + store: env.BLOBS, + }).pipe(Effect.ensuring(executor.close().pipe(Effect.ignore))); + }).pipe( Effect.catchCause((cause) => Effect.logWarning("internal integration reconcile aborted", cause).pipe( Effect.annotateLogs({ step: "reconcile-internal-integrations" }), @@ -56,29 +88,6 @@ export const reconcileInternalIntegrationsOnce = (env: CloudflareEnv): Promise (config: CloudflareConfig) => { - if (config.internalIntegrations.length === 0) return Effect.void; - return Effect.gen(function* () { - const accountId = internalPrincipal(config, { subject: null, email: null })?.accountId; - if (accountId === undefined) return; - yield* Effect.promise(() => preloadQuickJs()); - const dbHandle = yield* Effect.promise(() => createD1ExecutorDb(env.DB, env.BLOBS)); - const executor = yield* makeScopedExecutor( - accountId, - config.organizationId, - config.organizationName, - { orgWrites: "allowed" }, - ).pipe(Effect.provide(dbProviderLayer(Effect.succeed(dbHandle)))); - yield* reconcileInternalIntegrations(executor, config.internalIntegrations, { - listTools: makeInternalListTools(config.internalHosts), - store: env.BLOBS, - }).pipe(Effect.ensuring(executor.close().pipe(Effect.ignore))); - }).pipe( - Effect.provide(makeCloudflarePluginsProvider(config)), - Effect.provide(makeCloudflareHostConfig(config)), - ); -}; - export const makeCloudflareApp = async ( env: CloudflareEnv, config: CloudflareConfig = loadConfig(env), @@ -91,7 +100,7 @@ export const makeCloudflareApp = async ( // 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 dbHandle = await createD1ExecutorDb(env.DB, env.BLOBS); + const handle = await openDb(env); const identityLayer = cloudflareAccessIdentityLayer(config); const mcpAgentHandler = makeCloudflareMcpAgentHandler(config); const approvalHandler = makeCloudflareApprovalHandler(config, env); @@ -100,7 +109,7 @@ export const makeCloudflareApp = async ( plugins, providers: { identity: identityLayer, - db: dbProviderLayer(Effect.succeed(dbHandle)), + db: dbProviderLayer(Effect.succeed(handle)), engine: { codeExecutor: CloudflareCodeExecutorProvider }, plugins: { provider: makeCloudflarePluginsProvider(config),