perf(host-cloudflare): reconcile internal integrations on a cron instead of at isolate boot

apps/host-cloudflare/src/app.ts

@@ -4,9 +4,9 @@ import { HttpEffect, HttpRouter } from "effect/unstable/http";
 import {
   dbProviderLayer,
   ExecutorApp,
-  type ExecutorDbHandle,
   makeScopedExecutor,
   textFailureStrategy,
+  type ExecutorDbHandle,
 } from "@executor-js/api/server";
 
 import { internalPrincipal } from "./auth/cloudflare-access";
@@ -18,6 +18,7 @@ import { createD1ExecutorDb } from "./db/d1";
 import { cloudflareAccessIdentityLayer } from "./auth/cloudflare-access";
 import {
   CloudflareCodeExecutorProvider,
+  makeCloudflareExecutionStackLayer,
   makeCloudflareHostConfig,
   makeCloudflarePluginsProvider,
 } from "./execution";
@@ -42,43 +43,50 @@ import { preloadQuickJs } from "./quickjs";
 // so the providers close over it instead of reading process.env.
 // ===========================================================================
 
-let internalIntegrationsReconciled: Promise<void> | null = null;
+// 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<ExecutorDbHandle> | 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<void> => {
-  if (config.internalIntegrations.length === 0) return Promise.resolve();
-  internalIntegrationsReconciled ??= Effect.runPromise(
+const openDb = (env: CloudflareEnv): Promise<ExecutorDbHandle> => {
+  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<void> =>
+  Effect.runPromise(
     Effect.gen(function* () {
+      const config = loadConfig(env);
       const accountId = internalPrincipal(config, { subject: null, email: null })?.accountId;
-      if (accountId === undefined) return;
+      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,
         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.provide(dbProviderLayer(Effect.succeed(dbHandle))),
-      Effect.provide(makeCloudflarePluginsProvider(config)),
-      Effect.provide(makeCloudflareHostConfig(config)),
       Effect.catchCause((cause) =>
         Effect.logWarning("internal integration reconcile aborted", cause).pipe(
           Effect.annotateLogs({ step: "reconcile-internal-integrations" }),
         ),
       ),
     ),
   );
-  return internalIntegrationsReconciled;
-};
 
 export const makeCloudflareApp = async (
   env: CloudflareEnv,
@@ -92,8 +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);
-  await reconcileOncePerIsolate(config, dbHandle, env);
+  const handle = await openDb(env);
   const identityLayer = cloudflareAccessIdentityLayer(config);
   const mcpAgentHandler = makeCloudflareMcpAgentHandler(config);
   const approvalHandler = makeCloudflareApprovalHandler(config, env);
@@ -102,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),

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<void>();
+        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<ListedTool>;
+          });
+
+        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<typeof makeBinding>, store: HashStore) =>
     Effect.gen(function* () {

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<ReadonlyArray<readonly [string, ReconcileOutcome]>> =>
-  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" },
   );

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>): 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<unknown>[]): ExecutionContext => {
+  const context = { waitUntil: (promise: Promise<unknown>) => 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<unknown>[] = [];
+
+    // 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);
+  });
+});

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<CloudflareEnv> {
 export default {
   fetch: (request: Request, env: CloudflareEnv, ctx: ExecutionContext): Promise<Response> =>
     serveRequest(request, env, ctx, resolvePublic),
+  scheduled: (
+    _controller: ScheduledController,
+    env: CloudflareEnv,
+    ctx: ExecutionContext,
+  ): void => {
+    ctx.waitUntil(reconcileInternalIntegrationsOnce(env));
+  },
 };

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 },