From c35eaf3736aebd67122e31e6a6b50b886e566ea5 Mon Sep 17 00:00:00 2001 From: Ian Pascoe Date: Fri, 25 Sep 2026 05:46:48 -0400 Subject: [PATCH 1/2] Narrow plugin storage prefix reads in the database and scope OpenAPI operation scans --- .changeset/scoped-operation-scan.md | 6 + packages/core/sdk/src/executor.ts | 15 +- packages/core/sdk/src/plugin-storage.test.ts | 103 ++++++ .../plugins/openapi/src/sdk/store.test.ts | 299 +++++++++++------- packages/plugins/openapi/src/sdk/store.ts | 50 ++- 5 files changed, 332 insertions(+), 141 deletions(-) create mode 100644 .changeset/scoped-operation-scan.md diff --git a/.changeset/scoped-operation-scan.md b/.changeset/scoped-operation-scan.md new file mode 100644 index 0000000000..c69e0c06b0 --- /dev/null +++ b/.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. diff --git a/packages/core/sdk/src/executor.ts b/packages/core/sdk/src/executor.ts index fdbc9b7671..3cf7003197 100644 --- a/packages/core/sdk/src/executor.ts +++ b/packages/core/sdk/src/executor.ts @@ -1537,14 +1537,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 +1761,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 +1837,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) => diff --git a/packages/core/sdk/src/plugin-storage.test.ts b/packages/core/sdk/src/plugin-storage.test.ts index 3cc1d0793e..7b94844340 100644 --- a/packages/core/sdk/src/plugin-storage.test.ts +++ b/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) => @@ -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]); + }), + ); }); diff --git a/packages/plugins/openapi/src/sdk/store.test.ts b/packages/plugins/openapi/src/sdk/store.test.ts index ef4a4ed2cf..2cd626401c 100644 --- a/packages/plugins/openapi/src/sdk/store.test.ts +++ b/packages/plugins/openapi/src/sdk/store.test.ts @@ -1,5 +1,5 @@ import { describe, expect, it } from "@effect/vitest"; -import { Effect, Option } from "effect"; +import { Effect, Option, Schema } from "effect"; import { Subject, @@ -10,140 +10,193 @@ import { type StorageDeps, } from "@executor-js/sdk/core"; -import { makeDefaultOpenapiStore } from "./store"; +import { makeDefaultOpenapiStore, type StoredOperation } from "./store"; import { OperationBinding } from "./types"; +const encodeBinding = Schema.encodeSync(OperationBinding); + +const makeStoreHarness = () => { + const rows = new Map(); + const capturedKeys: string[] = []; + const listedPrefixes: (string | undefined)[] = []; + const storageKey = (collection: string, key: string) => `${collection}\0${key}`; + const now = new Date(); + const makeEntry = (input: { + readonly owner: "org" | "user"; + readonly collection: string; + readonly key: string; + readonly data: T; + }): PluginStorageEntry => ({ + id: storageKey(input.collection, input.key), + owner: input.owner, + pluginId: "openapi", + collection: input.collection, + key: input.key, + data: input.data, + createdAt: now, + updatedAt: now, + }); + const pluginStorage: PluginStorageFacade = { + collection: () => ({ + get: () => Effect.succeed(null), + getForOwner: () => Effect.succeed(null), + list: () => Effect.succeed([]), + put: (input) => + Effect.succeed( + makeEntry({ + owner: input.owner, + collection: "unused", + key: input.key, + data: input.data, + }), + ), + query: () => Effect.succeed([]), + count: () => Effect.succeed(0), + remove: () => Effect.void, + }), + get: (input: { readonly collection: string; readonly key: string }) => + Effect.succeed( + (rows.get(storageKey(input.collection, input.key)) as PluginStorageEntry | undefined) ?? + null, + ), + getForOwner: (input: { readonly collection: string; readonly key: string }) => + Effect.succeed( + (rows.get(storageKey(input.collection, input.key)) as PluginStorageEntry | undefined) ?? + null, + ), + list: (input: { readonly collection: string; readonly keyPrefix?: string }) => + Effect.sync(() => { + listedPrefixes.push(input.keyPrefix); + return [...rows.values()].filter( + (row) => + row.collection === input.collection && + (input.keyPrefix === undefined || row.key.startsWith(input.keyPrefix)), + ) as PluginStorageEntry[]; + }), + put: (input: { + readonly owner: "org" | "user"; + readonly collection: string; + readonly key: string; + readonly data: unknown; + }) => { + const entry = makeEntry({ ...input, data: input.data as T }); + rows.set(storageKey(input.collection, input.key), entry); + return Effect.succeed(entry); + }, + putMany: (input) => + Effect.sync(() => { + for (const entry of input.entries) { + capturedKeys.push(entry.key); + rows.set( + storageKey(entry.collection, entry.key), + makeEntry({ + owner: input.owner, + collection: entry.collection, + key: entry.key, + data: entry.data, + }), + ); + } + }), + remove: (input) => + Effect.sync(() => { + rows.delete(storageKey(input.collection, input.key)); + }), + removeMany: (input) => + Effect.sync(() => { + for (const entry of input.entries) { + rows.delete(storageKey(entry.collection, entry.key)); + } + }), + }; + const blobs: PluginBlobStore = { + get: () => Effect.succeed(null), + put: () => Effect.void, + delete: () => Effect.void, + has: () => Effect.succeed(false), + }; + const store = makeDefaultOpenapiStore({ + owner: { tenant: Tenant.make("tenant"), subject: Subject.make("subject") }, + blobs, + pluginStorage, + } satisfies StorageDeps); + return { store, pluginStorage, capturedKeys, listedPrefixes }; +}; + +const operation = (integration: string, toolName: string): StoredOperation => ({ + integration, + toolName, + binding: OperationBinding.make({ + method: "get", + servers: [], + pathTemplate: `/${toolName}`, + parameters: [], + requestBody: Option.none(), + responseBody: Option.none(), + }), +}); + describe("OpenAPI operation store", () => { it.effect("bounds operation storage keys while preserving tool-name lookup", () => Effect.gen(function* () { - const rows = new Map(); - const capturedKeys: string[] = []; - const storageKey = (collection: string, key: string) => `${collection}\0${key}`; - const now = new Date(); - const makeEntry = (input: { - readonly owner: "org" | "user"; - readonly collection: string; - readonly key: string; - readonly data: T; - }): PluginStorageEntry => ({ - id: storageKey(input.collection, input.key), - owner: input.owner, - pluginId: "openapi", - collection: input.collection, - key: input.key, - data: input.data, - createdAt: now, - updatedAt: now, - }); - const pluginStorage: PluginStorageFacade = { - collection: () => ({ - get: () => Effect.succeed(null), - getForOwner: () => Effect.succeed(null), - list: () => Effect.succeed([]), - put: (input) => - Effect.succeed( - makeEntry({ - owner: input.owner, - collection: "unused", - key: input.key, - data: input.data, - }), - ), - query: () => Effect.succeed([]), - count: () => Effect.succeed(0), - remove: () => Effect.void, - }), - get: (input: { readonly collection: string; readonly key: string }) => - Effect.succeed( - (rows.get(storageKey(input.collection, input.key)) as - | PluginStorageEntry - | undefined) ?? null, - ), - getForOwner: (input: { readonly collection: string; readonly key: string }) => - Effect.succeed( - (rows.get(storageKey(input.collection, input.key)) as - | PluginStorageEntry - | undefined) ?? null, - ), - list: (input: { readonly collection: string; readonly keyPrefix?: string }) => - Effect.succeed( - [...rows.values()].filter( - (row) => - row.collection === input.collection && - (input.keyPrefix === undefined || row.key.startsWith(input.keyPrefix)), - ) as PluginStorageEntry[], - ), - put: (input: { - readonly owner: "org" | "user"; - readonly collection: string; - readonly key: string; - readonly data: unknown; - }) => { - const entry = makeEntry({ ...input, data: input.data as T }); - rows.set(storageKey(input.collection, input.key), entry); - return Effect.succeed(entry); - }, - putMany: (input) => - Effect.sync(() => { - for (const entry of input.entries) { - capturedKeys.push(entry.key); - rows.set( - storageKey(entry.collection, entry.key), - makeEntry({ - owner: input.owner, - collection: entry.collection, - key: entry.key, - data: entry.data, - }), - ); - } - }), - remove: (input) => - Effect.sync(() => { - rows.delete(storageKey(input.collection, input.key)); - }), - removeMany: (input) => - Effect.sync(() => { - for (const entry of input.entries) { - rows.delete(storageKey(entry.collection, entry.key)); - } - }), - }; - const blobs: PluginBlobStore = { - get: () => Effect.succeed(null), - put: () => Effect.void, - delete: () => Effect.void, - has: () => Effect.succeed(false), - }; - const store = makeDefaultOpenapiStore({ - owner: { tenant: Tenant.make("tenant"), subject: Subject.make("subject") }, - blobs, - pluginStorage, - } satisfies StorageDeps); + const { store, capturedKeys } = makeStoreHarness(); const toolName = `users.${"veryLongSegment.".repeat(40)}get`; - yield* store.putOperations("microsoft_graph", [ - { - integration: "microsoft_graph", - toolName, - binding: OperationBinding.make({ - method: "get", - servers: [], - pathTemplate: "/users/{userId}/messages", - parameters: [], - requestBody: Option.none(), - responseBody: Option.none(), - }), - }, - ]); + yield* store.putOperations("microsoft_graph", [operation("microsoft_graph", toolName)]); expect(capturedKeys).toHaveLength(1); expect(capturedKeys[0]!.length).toBeLessThanOrEqual(255); expect(capturedKeys[0]).not.toContain(toolName); - const operation = yield* store.getOperation("microsoft_graph", toolName); - expect(operation?.toolName).toBe(toolName); - expect(operation?.binding.pathTemplate).toBe("/users/{userId}/messages"); + const stored = yield* store.getOperation("microsoft_graph", toolName); + expect(stored?.toolName).toBe(toolName); + expect(stored?.binding.pathTemplate).toBe(`/${toolName}`); + }), + ); + + it.effect("lists and removes one integration's operations without reading the others", () => + Effect.gen(function* () { + const { store, pluginStorage, listedPrefixes } = makeStoreHarness(); + yield* store.putOperations("github", [ + operation("github", "repos.get"), + operation("github", "issues.list"), + ]); + yield* store.putOperations("stripe", [operation("stripe", "customers.list")]); + const legacyRow = (key: string, integration: string, toolName: string) => + pluginStorage.put({ + owner: "org", + collection: "operation", + key, + data: { + integration, + toolName, + binding: encodeBinding(operation(integration, toolName).binding), + }, + }); + // A row written under the legacy `.` key scheme. + yield* legacyRow("github.pulls.list", "github", "pulls.list"); + // A legacy key that starts with `github.` but belongs to another + // integration: the prefix over-matches, the result must not. + yield* legacyRow("github.enterprise.repos.get", "github.enterprise", "repos.get"); + + listedPrefixes.length = 0; + const github = yield* store.listOperations("github"); + expect(github.map((entry) => entry.toolName).sort()).toEqual([ + "issues.list", + "pulls.list", + "repos.get", + ]); + expect(github.every((entry) => entry.integration === "github")).toBe(true); + expect(listedPrefixes).not.toContain(undefined); + + yield* store.removeOperations("github"); + expect(yield* store.listOperations("github")).toEqual([]); + expect((yield* store.listOperations("stripe")).map((entry) => entry.toolName)).toEqual([ + "customers.list", + ]); + expect( + (yield* store.listOperations("github.enterprise")).map((entry) => entry.toolName), + ).toEqual(["repos.get"]); }), ); }); diff --git a/packages/plugins/openapi/src/sdk/store.ts b/packages/plugins/openapi/src/sdk/store.ts index 548a521031..cae0ba8918 100644 --- a/packages/plugins/openapi/src/sdk/store.ts +++ b/packages/plugins/openapi/src/sdk/store.ts @@ -1,4 +1,4 @@ -import { Effect, Option, Predicate, Schema } from "effect"; +import { Effect, Option, Schema } from "effect"; import { type PluginStorageEntry, @@ -84,11 +84,17 @@ const stableKeyHash = (value: string): string => { return hash.toString(36).padStart(13, "0"); }; +/** Every current-scheme key for an integration starts with this. */ +const operationKeyPrefix = (integration: string): string => + `${OPERATION_KEY_VERSION}.${stableKeyHash(integration)}.`; + const operationKey = (integration: string, toolName: string): string => - `${OPERATION_KEY_VERSION}.${stableKeyHash(integration)}.${stableKeyHash(toolName)}`; + `${operationKeyPrefix(integration)}${stableKeyHash(toolName)}`; + +const legacyOperationKeyPrefix = (integration: string): string => `${integration}.`; const legacyOperationKey = (integration: string, toolName: string): string => - `${integration}.${toolName}`; + `${legacyOperationKeyPrefix(integration)}${toolName}`; /** Blob key for a spec's content hash. Content-addressed so re-puts are * idempotent and identical specs share one blob per partition. */ @@ -147,21 +153,35 @@ export const makeDefaultOpenapiStore = ({ pluginStorage, blobs }: StorageDeps): ...(operation.description !== undefined ? { description: operation.description } : {}), }); - const listRows = (integration: string) => - pluginStorage - .list({ collection: OPERATION_COLLECTION }) - .pipe( - Effect.map((rows: readonly PluginStorageEntry[]) => - rows.filter((row) => rowToOperation(row)?.integration === integration), - ), - ); + // Reads only this integration's rows: both key schemes carry the + // integration as a prefix, so storage narrows the read instead of loading + // every integration's operations into memory. Each row is decoded once; the + // `integration` check drops prefix over-matches (a hash-prefix collision, or + // a legacy `.` prefix that is also a prefix of another slug's + // keys) so the result is exact. + const listEntries = (integration: string) => + Effect.gen(function* () { + const prefixes = [operationKeyPrefix(integration), legacyOperationKeyPrefix(integration)]; + const seen = new Set(); + const entries: { readonly key: string; readonly operation: StoredOperation }[] = []; + for (const keyPrefix of prefixes) { + const rows = yield* pluginStorage.list({ collection: OPERATION_COLLECTION, keyPrefix }); + for (const row of rows) { + if (seen.has(row.key)) continue; + seen.add(row.key); + const operation = rowToOperation(row); + if (operation?.integration === integration) entries.push({ key: row.key, operation }); + } + } + return entries; + }); const removeOperations = (integration: string) => Effect.gen(function* () { - const rows = yield* listRows(integration); + const entries = yield* listEntries(integration); yield* pluginStorage.removeMany({ owner: STORE_OWNER, - entries: rows.map((row) => ({ collection: OPERATION_COLLECTION, key: row.key })), + entries: entries.map((entry) => ({ collection: OPERATION_COLLECTION, key: entry.key })), }); }); @@ -201,8 +221,8 @@ export const makeDefaultOpenapiStore = ({ pluginStorage, blobs }: StorageDeps): }), listOperations: (integration) => - listRows(integration).pipe( - Effect.map((rows) => rows.map(rowToOperation).filter(Predicate.isNotNull)), + listEntries(integration).pipe( + Effect.map((entries) => entries.map((entry) => entry.operation)), ), removeOperations, From e94b935b7684d4086ec00c9b99394b8637d37724 Mon Sep 17 00:00:00 2001 From: Ian Pascoe Date: Fri, 25 Sep 2026 05:50:48 -0400 Subject: [PATCH 2/2] Let hosts bound stale tool-catalog rebuild concurrency; Cloudflare rebuilds two at a time --- .changeset/tools-sync-concurrency.md | 5 + apps/host-cloudflare/src/execution.ts | 8 ++ .../core/api/src/server/scoped-executor.ts | 10 ++ packages/core/sdk/src/executor.ts | 20 ++- .../plugins/mcp/src/sdk/catalog-sync.test.ts | 134 ++++++++++-------- 5 files changed, 110 insertions(+), 67 deletions(-) create mode 100644 .changeset/tools-sync-concurrency.md diff --git a/.changeset/tools-sync-concurrency.md b/.changeset/tools-sync-concurrency.md new file mode 100644 index 0000000000..68d711d0ba --- /dev/null +++ b/.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. diff --git a/apps/host-cloudflare/src/execution.ts b/apps/host-cloudflare/src/execution.ts index 1ee0e4b1e9..52e2c39df1 100644 --- a/apps/host-cloudflare/src/execution.ts +++ b/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 => Layer.succeed(HostConfig)({ allowLocalNetwork: config.allowLocalNetwork, webBaseUrl: config.webBaseUrl, oauthCallbackPath: "/api/oauth/callback", + toolsSyncConcurrency: CLOUDFLARE_TOOLS_SYNC_CONCURRENCY, }); /** diff --git a/packages/core/api/src/server/scoped-executor.ts b/packages/core/api/src/server/scoped-executor.ts index 749839be3e..52e07634ff 100644 --- a/packages/core/api/src/server/scoped-executor.ts +++ b/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 }), diff --git a/packages/core/sdk/src/executor.ts b/packages/core/sdk/src/executor.ts index 3cf7003197..6343dd39ad 100644 --- a/packages/core/sdk/src/executor.ts +++ b/packages/core/sdk/src/executor.ts @@ -795,6 +795,16 @@ export interface ExecutorConfig 0) { const background = yield* Effect.forkDetach( - Effect.all(deferred, { concurrency: STALE_TOOLS_SYNC_CONCURRENCY }), + Effect.all(deferred, { concurrency: toolsSyncConcurrency }), ); config.waitUntil?.( new Promise((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 diff --git a/packages/plugins/mcp/src/sdk/catalog-sync.test.ts b/packages/plugins/mcp/src/sdk/catalog-sync.test.ts index 34ec9f24f8..5fd5020fe1 100644 --- a/packages/plugins/mcp/src/sdk/catalog-sync.test.ts +++ b/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 }), ); });