Let hosts bound stale tool-catalog rebuild concurrency; Cloudflare rebuilds two at a time

Author: ian-pascoe

.changeset/tools-sync-concurrency.md

@@ -0,0 +1,5 @@
+---
+"@executor-js/sdk": patch
+---
+
+`ExecutorConfig.toolsSyncConcurrency` sets how many stale tool catalogs one tools read rebuilds at once (default 10, unchanged). Each in-flight rebuild holds its resolved catalog in memory until its write commits, so memory-constrained hosts can narrow the fan-out; the Cloudflare host now rebuilds two at a time, keeping a full stale fan-out over large OpenAPI specs inside the Workers isolate limit.

apps/host-cloudflare/src/execution.ts

@@ -51,11 +51,19 @@ export const makeCloudflarePluginsProvider = (
       }),
   });
 
+// Two stale catalogs rebuild at once, not the SDK's ten: each in-flight
+// rebuild holds its resolved tools and schema definitions until its write
+// commits, and a full fan-out over a few large OpenAPI specs overruns the
+// 128MB Workers isolate. Writes are serialized anyway, so the narrower
+// fan-out costs little convergence time.
+const CLOUDFLARE_TOOLS_SYNC_CONCURRENCY = 2;
+
 export const makeCloudflareHostConfig = (config: CloudflareConfig): Layer.Layer<HostConfig> =>
   Layer.succeed(HostConfig)({
     allowLocalNetwork: config.allowLocalNetwork,
     webBaseUrl: config.webBaseUrl,
     oauthCallbackPath: "/api/oauth/callback",
+    toolsSyncConcurrency: CLOUDFLARE_TOOLS_SYNC_CONCURRENCY,
   });
 
 /**

packages/core/api/src/server/scoped-executor.ts

@@ -126,6 +126,13 @@ export interface HostConfigShape {
    * operator knob.
    */
   readonly toolsSyncTtlMs?: number | null;
+  /**
+   * Forwarded verbatim to `ExecutorConfig.toolsSyncConcurrency`: how many
+   * stale tool catalogs one read rebuilds at once. Omit to take the SDK
+   * default; memory-constrained hosts lower it because every in-flight
+   * rebuild holds its resolved catalog until its write commits.
+   */
+  readonly toolsSyncConcurrency?: number;
   /**
    * Forwarded to `ExecutorConfig.waitUntil`: the host's keep-alive
    * for background work that outlives a request (stale tool-catalog rebuilds
@@ -334,6 +341,9 @@ export const makeScopedExecutor = <
       fetch: hostedFetch,
       onIntegrationChange: config.onIntegrationChange,
       ...(config.toolsSyncTtlMs !== undefined ? { toolsSyncTtlMs: config.toolsSyncTtlMs } : {}),
+      ...(config.toolsSyncConcurrency !== undefined
+        ? { toolsSyncConcurrency: config.toolsSyncConcurrency }
+        : {}),
       ...(waitUntil !== undefined ? { waitUntil } : {}),
       onElicitation: "accept-all",
       ...(options?.orgWrites === undefined ? {} : { orgWrites: options.orgWrites }),

packages/core/sdk/src/executor.ts

@@ -795,6 +795,16 @@ export interface ExecutorConfig<TPlugins extends readonly AnyPlugin[] = readonly
    * mode: a read then always reflects a fully converged catalog).
    */
   readonly toolsSyncGraceMs?: number | null;
+  /**
+   * How many stale connection catalogs one tools read rebuilds at once.
+   * Defaults to {@link STALE_TOOLS_SYNC_CONCURRENCY}. Every in-flight rebuild
+   * holds its connection's resolved tool set and schema definitions in memory
+   * until its catalog write gets the single persist permit, so hosts with a
+   * small memory ceiling (Cloudflare Workers' 128MB isolate) lower it; hosts
+   * whose catalogs mostly come from slow remote listings keep the default so
+   * those listings overlap.
+   */
+  readonly toolsSyncConcurrency?: number;
   /**
    * Host keep-alive for background work that outlives a request — the
    * platform `waitUntil` on Cloudflare Workers, where I/O started inside a
@@ -5546,6 +5556,10 @@ export const createExecutor = <const TPlugins extends readonly AnyPlugin[] = rea
     const toolsSyncTtlMs =
       config.toolsSyncTtlMs === undefined ? DEFAULT_TOOLS_SYNC_TTL_MS : config.toolsSyncTtlMs;
 
+    // How many stale catalogs a tools read rebuilds at once
+    // (`ExecutorConfig.toolsSyncConcurrency`).
+    const toolsSyncConcurrency = config.toolsSyncConcurrency ?? STALE_TOOLS_SYNC_CONCURRENCY;
+
     // Rebuild any visible connection whose persisted tool catalog is stale.
     // Three triggers:
     //  - stale-marked: `tools_synced_at` is NULL (`connections.markToolsStale`
@@ -5681,15 +5695,13 @@ export const createExecutor = <const TPlugins extends readonly AnyPlugin[] = rea
         }
         if (deferred.length > 0) {
           const background = yield* Effect.forkDetach(
-            Effect.all(deferred, { concurrency: STALE_TOOLS_SYNC_CONCURRENCY }),
+            Effect.all(deferred, { concurrency: toolsSyncConcurrency }),
           );
           config.waitUntil?.(
             new Promise<void>((resolve) => background.addObserver(() => resolve(undefined))),
           );
         }
-        yield* Effect.all(urgent, {
-          concurrency: STALE_TOOLS_SYNC_CONCURRENCY,
-        });
+        yield* Effect.all(urgent, { concurrency: toolsSyncConcurrency });
       });
 
     // How long a tools read waits for the stale sync before answering from

packages/plugins/mcp/src/sdk/catalog-sync.test.ts

@@ -297,15 +297,13 @@ describe("MCP tools/list pagination", () => {
 //
 // The fixture below refuses to answer any listing until the bound is reached,
 // which pins both edges at once: a serial refresh parks on the first listing
-// and never finishes, while an unbounded refresh puts more than
-// STALE_TOOLS_SYNC_CONCURRENCY listings in flight. The stale set is deliberately
-// one larger than the bound, so the last connection can only be served after an
-// earlier one completes.
+// and never finishes, while an unbounded refresh puts more than the bound
+// (STALE_TOOLS_SYNC_CONCURRENCY, or the host's `toolsSyncConcurrency`) in
+// flight. The stale set is deliberately one larger than the bound, so the last
+// connection can only be served after an earlier one completes.
 // ---------------------------------------------------------------------------
 
-const STALE_CONNECTIONS = STALE_TOOLS_SYNC_CONCURRENCY + 1;
-
-const serveLatchedListServer = () =>
+const serveLatchedListServer = (bound: number) =>
   Effect.gen(function* () {
     const armed = yield* Ref.make(false);
     const listings = yield* Ref.make(0);
@@ -341,7 +339,7 @@ const serveLatchedListServer = () =>
         // refresh parks on the first one and never reaches the bound.
         if (yield* Ref.get(armed)) {
           const arrived = yield* Ref.updateAndGet(listings, (n) => n + 1);
-          if (arrived >= STALE_TOOLS_SYNC_CONCURRENCY) {
+          if (arrived >= bound) {
             yield* Deferred.succeed(atLimit, undefined);
           }
           yield* Deferred.await(release);
@@ -361,67 +359,77 @@ const serveLatchedListServer = () =>
     } as const;
   });
 
+const expectBoundedStaleRefresh = (options: { readonly toolsSyncConcurrency?: number }) =>
+  Effect.gen(function* () {
+    const bound = options.toolsSyncConcurrency ?? STALE_TOOLS_SYNC_CONCURRENCY;
+    const staleConnections = bound + 1;
+    const fixture = yield* serveLatchedListServer(bound);
+    const executor = yield* createExecutor({
+      ...makeTestConfig({ plugins: [memoryCredentialsPlugin(), mcpPlugin()] as const }),
+      // Everything is expired on every read, so a single tools read has the
+      // whole set to rebuild.
+      toolsSyncTtlMs: 0,
+      // Strict mode: the assertions below synchronize on the read fiber
+      // completing only after every rebuild has finished. With a grace
+      // budget the read would return early and `Fiber.join` would no longer
+      // order the final listing before the count assertion.
+      toolsSyncGraceMs: null,
+      ...options,
+    });
+
+    for (let index = 0; index < staleConnections; index++) {
+      const slug = IntegrationSlug.make(`latched_mcp_${index}`);
+      yield* executor.mcp.addServer({
+        name: `latched-mcp-${index}`,
+        endpoint: fixture.endpoint(index),
+        slug: String(slug),
+      });
+      yield* executor.connections.create({
+        owner: "org",
+        name: CONNECTION,
+        integration: slug,
+        template: TEMPLATE,
+        value: "",
+      });
+    }
+
+    // Warm every catalog while the fixture still answers freely, so the
+    // latched read below is purely the stale-refresh fan-out.
+    yield* executor.tools.list();
+    yield* fixture.arm;
+
+    const readFiber = yield* Effect.forkChild(executor.tools.list());
+
+    // Timeouts are well inside the harness limit, so a broken fan-out fails
+    // on an assertion here rather than as an opaque test-runner timeout.
+    // A serial refresh never saturates the bound and fails on this line.
+    const saturated = yield* fixture.awaitLimit.pipe(Effect.timeoutOption("10 seconds"));
+    expect(Option.isSome(saturated)).toBe(true);
+
+    // The bound is reached and every one of those listings is still parked.
+    // Give an unbounded fan-out ample time to dial the remaining connection:
+    // it never may, because no permit has been given back yet.
+    yield* Effect.sleep("500 millis");
+    expect(yield* fixture.listings).toBe(bound);
+
+    // Releasing the parked listings frees permits, and only then does the
+    // last connection get dialled.
+    yield* fixture.release;
+    const refreshed = yield* Fiber.join(readFiber).pipe(Effect.timeoutOption("10 seconds"));
+    expect(Option.isSome(refreshed)).toBe(true);
+    expect(yield* fixture.listings).toBe(staleConnections);
+  });
+
 describe("MCP stale-catalog refresh", () => {
   // `it.live` (real clock): proving that nothing beyond the bound is dialled
   // means giving a real HTTP round trip a real window to happen in, and the
   // timeouts below must actually fire. The TestClock advances neither.
   it.live("rebuilds stale connections concurrently up to the bound, then queues the rest", () =>
-    Effect.gen(function* () {
-      const fixture = yield* serveLatchedListServer();
-      const executor = yield* createExecutor({
-        ...makeTestConfig({ plugins: [memoryCredentialsPlugin(), mcpPlugin()] as const }),
-        // Everything is expired on every read, so a single tools read has the
-        // whole set to rebuild.
-        toolsSyncTtlMs: 0,
-        // Strict mode: the assertions below synchronize on the read fiber
-        // completing only after every rebuild has finished. With a grace
-        // budget the read would return early and `Fiber.join` would no longer
-        // order the final listing before the count assertion.
-        toolsSyncGraceMs: null,
-      });
-
-      for (let index = 0; index < STALE_CONNECTIONS; index++) {
-        const slug = IntegrationSlug.make(`latched_mcp_${index}`);
-        yield* executor.mcp.addServer({
-          name: `latched-mcp-${index}`,
-          endpoint: fixture.endpoint(index),
-          slug: String(slug),
-        });
-        yield* executor.connections.create({
-          owner: "org",
-          name: CONNECTION,
-          integration: slug,
-          template: TEMPLATE,
-          value: "",
-        });
-      }
-
-      // Warm every catalog while the fixture still answers freely, so the
-      // latched read below is purely the stale-refresh fan-out.
-      yield* executor.tools.list();
-      yield* fixture.arm;
-
-      const readFiber = yield* Effect.forkChild(executor.tools.list());
-
-      // Timeouts are well inside the harness limit, so a broken fan-out fails
-      // on an assertion here rather than as an opaque test-runner timeout.
-      // A serial refresh never saturates the bound and fails on this line.
-      const saturated = yield* fixture.awaitLimit.pipe(Effect.timeoutOption("10 seconds"));
-      expect(Option.isSome(saturated)).toBe(true);
-
-      // The bound is reached and every one of those listings is still parked.
-      // Give an unbounded fan-out ample time to dial the remaining connection:
-      // it never may, because no permit has been given back yet.
-      yield* Effect.sleep("500 millis");
-      expect(yield* fixture.listings).toBe(STALE_TOOLS_SYNC_CONCURRENCY);
+    expectBoundedStaleRefresh({}),
+  );
 
-      // Releasing the parked listings frees permits, and only then does the
-      // last connection get dialled.
-      yield* fixture.release;
-      const refreshed = yield* Fiber.join(readFiber).pipe(Effect.timeoutOption("10 seconds"));
-      expect(Option.isSome(refreshed)).toBe(true);
-      expect(yield* fixture.listings).toBe(STALE_CONNECTIONS);
-    }),
+  it.live("bounds the stale rebuild fan-out at the host's toolsSyncConcurrency", () =>
+    expectBoundedStaleRefresh({ toolsSyncConcurrency: 2 }),
   );
 });