Narrow plugin storage prefix reads in SQL and bound stale catalog rebuild concurrency

.changeset/scoped-operation-scan.md

@@ -0,0 +1,6 @@
+---
+"@executor-js/sdk": patch
+"@executor-js/plugin-openapi": patch
+---
+
+Plugin storage key-prefix reads narrow in the database instead of loading the whole collection and filtering in memory. OpenAPI catalog rebuilds now read only the rebuilt integration's operations, decoding each once, where they previously loaded every OpenAPI integration's operations for each connection — the allocation that pushed Cloudflare-hosted sessions with large specs (for example Cloudflare's own API) past the Workers memory limit during tool search.

.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
@@ -1537,14 +1547,23 @@ const makePluginStorageFacade = (input: {
   const tenant = String(input.owner.tenant);
 
   const whereFor =
-    (collection: string, key?: string): CoreWhere =>
+    (collection: string, key?: string, keyPrefix?: string): CoreWhere =>
     (b: AnyCb) =>
       b.and(
         b("plugin_id", "=", input.pluginId),
         b("collection", "=", collection),
         key === undefined ? true : b("key", "=", key),
+        keyPrefix === undefined ? true : b("key", "starts with", keyPrefix),
       );
 
+  // `starts with` compiles to an unescaped LIKE on SQL adapters (and a
+  // case-insensitive one on SQLite), so the pushed-down prefix only narrows
+  // the read to a superset; `list` still applies the exact `startsWith`. A
+  // backslash is Postgres LIKE's default escape character and could turn the
+  // superset into a subset, so such prefixes are filtered in JS only.
+  const sqlKeyPrefix = (keyPrefix: string | undefined): string | undefined =>
+    keyPrefix === undefined || keyPrefix.includes("\\") ? undefined : keyPrefix;
+
   const whereOwner = (owner: Owner, collection: string, key: string): CoreWhere => {
     const os = ownerSubject(owner);
     return (b: AnyCb) =>
@@ -1752,7 +1771,7 @@ const makePluginStorageFacade = (input: {
       if (validationError) return yield* validationError;
 
       const rows = yield* input.core.findMany("plugin_storage", {
-        where: whereFor(definition.name),
+        where: whereFor(definition.name, undefined, sqlKeyPrefix(queryInput?.keyPrefix)),
       });
       const filtered = sortByOwnerPrecedence(rows)
         .filter((row) =>
@@ -1828,7 +1847,7 @@ const makePluginStorageFacade = (input: {
     list: (storageInput) =>
       Effect.gen(function* () {
         const rows = yield* input.core.findMany("plugin_storage", {
-          where: whereFor(storageInput.collection),
+          where: whereFor(storageInput.collection, undefined, sqlKeyPrefix(storageInput.keyPrefix)),
         });
         return sortByOwnerPrecedence(rows)
           .filter((row) =>
@@ -5537,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`
@@ -5672,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/core/sdk/src/plugin-storage.test.ts

@@ -79,6 +79,8 @@ const executionHistoryPlugin = definePlugin(() => ({
         owner,
         entries: keys.map((key) => ({ collection: toolCalls.name, key })),
       }),
+    listByPrefix: (keyPrefix: string) =>
+      ctx.pluginStorage.list({ collection: toolCalls.name, keyPrefix }),
     get: (key: string) => ctx.storage.toolCalls.get({ key }),
     getForOwner: (owner: Owner, key: string) => ctx.storage.toolCalls.getForOwner({ owner, key }),
     query: (input?: PluginStorageCollectionQueryInput<typeof toolCalls>) =>
@@ -163,6 +165,39 @@ const failPluginStorageBulkWriteAfterFirstRow = (db: FumaDb): FumaDb => {
   return wrap(db);
 };
 
+// Records how many `plugin_storage` rows each adapter read hands back, so a
+// test can tell a prefix applied in SQL from one applied after loading every
+// row of the collection into memory.
+const countPluginStorageReads = (db: FumaDb, rowCounts: number[]): FumaDb => {
+  const wrap = (source: FumaDb): FumaDb =>
+    new Proxy(source, {
+      get(target, property, receiver) {
+        if (property === "withContext") {
+          const withContext = target.withContext;
+          return withContext === undefined
+            ? undefined
+            : (context: unknown) => wrap(withContext(context));
+        }
+        if (property === "transaction") {
+          const transaction: FumaDb["transaction"] = (run) =>
+            target.transaction((transactionDb) => run(wrap(transactionDb)));
+          return transaction;
+        }
+        if (property === "findMany") {
+          const findMany: FumaDb["findMany"] = async (table, options) => {
+            const rows = await target.findMany(table, options);
+            if (table === "plugin_storage") rowCounts.push(rows.length);
+            return rows;
+          };
+          return findMany;
+        }
+        return Reflect.get(target, property, receiver);
+      },
+    });
+
+  return wrap(db);
+};
+
 describe("plugin storage collections", () => {
   it.effect("queries declared indexes through the executor's SQLite FumaDB target", () =>
     Effect.gen(function* () {
@@ -524,4 +559,72 @@ describe("plugin storage collections", () => {
       });
     }),
   );
+
+  it.effect("narrows key-prefix reads in storage and keeps the result exact", () =>
+    Effect.gen(function* () {
+      const config = makeTestConfig({
+        backend: "sqlite",
+        plugins: [executionHistoryPlugin] as const,
+      });
+      const rowCounts: number[] = [];
+      const executor = yield* Effect.acquireRelease(
+        createExecutor({ ...config, db: countPluginStorageReads(config.db, rowCounts) }),
+        (instance) =>
+          instance
+            .close()
+            .pipe(
+              Effect.ignore,
+              Effect.andThen(Effect.promise(() => config.testDb.close()).pipe(Effect.ignore)),
+            ),
+      );
+
+      const keys = [
+        "op.abc.1",
+        "op.abc.2",
+        "op.abd.3",
+        // `_` and `%` are LIKE wildcards and SQLite LIKE ignores ASCII case, so
+        // each exact key below has look-alikes a naive pushdown would return.
+        "cloudflare_com.a",
+        "cloudflareXcom.b",
+        "CLOUDFLARE_COM.c",
+        "cloudflare%com.d",
+        "cloudflare-com.e",
+        // A backslash is Postgres LIKE's default escape character.
+        "back\\slash.f",
+        "backslash.g",
+        ...Array.from({ length: 40 }, (_, index) => `filler-${String(index).padStart(2, "0")}`),
+      ];
+      yield* executor.executionHistory.recordMany(
+        "org",
+        keys.map((key, index) => ({
+          key,
+          data: call({
+            runId: "run-prefix",
+            toolId: key,
+            status: "ok",
+            startedAt: new Date(Date.UTC(2026, 4, 29, 13, 0, index)).toISOString(),
+          }),
+        })),
+      );
+
+      const listed = (keyPrefix: string) =>
+        executor.executionHistory
+          .listByPrefix(keyPrefix)
+          .pipe(Effect.map((rows) => rows.map((row) => row.key).sort()));
+
+      rowCounts.length = 0;
+      expect(yield* listed("op.abc.")).toEqual(["op.abc.1", "op.abc.2"]);
+      // Only the matching rows left storage; the other 48 were never loaded.
+      expect(rowCounts).toEqual([2]);
+
+      expect(yield* listed("cloudflare_com.")).toEqual(["cloudflare_com.a"]);
+      expect(yield* listed("cloudflare%com.")).toEqual(["cloudflare%com.d"]);
+      expect(yield* listed("back\\slash.")).toEqual(["back\\slash.f"]);
+
+      rowCounts.length = 0;
+      const queried = yield* executor.executionHistory.query({ keyPrefix: "op.abc." });
+      expect(queried.map((entry) => entry.key).sort()).toEqual(["op.abc.1", "op.abc.2"]);
+      expect(rowCounts).toEqual([2]);
+    }),
+  );
 });