From ea6557c297536e8c363e90730a5c80517675ea81 Mon Sep 17 00:00:00 2001 From: Ian Pascoe Date: Mon, 21 Sep 2026 06:14:23 -0400 Subject: [PATCH 1/4] Report the Cloudflare Access principal on /account/members The console infers workspace-admin from that list, so an empty response locked edit/delete even when ADMIN_EMAILS granted admin. --- .../src/account/account-provider.test.ts | 54 +++++++++++++++++++ .../src/account/account-provider.ts | 31 +++++++++-- 2 files changed, 80 insertions(+), 5 deletions(-) create mode 100644 apps/host-cloudflare/src/account/account-provider.test.ts diff --git a/apps/host-cloudflare/src/account/account-provider.test.ts b/apps/host-cloudflare/src/account/account-provider.test.ts new file mode 100644 index 0000000000..de0c8bb8f8 --- /dev/null +++ b/apps/host-cloudflare/src/account/account-provider.test.ts @@ -0,0 +1,54 @@ +import { describe, expect, it } from "@effect/vitest"; +import { Effect } from "effect"; + +import { AccountUnauthorized } from "@executor-js/api"; +import { AccountProvider } from "@executor-js/api/server"; + +import type { CloudflareConfig } from "../config"; +import { cloudflareAccountProvider } from "./account-provider"; + +const config = (overrides: Partial = {}): CloudflareConfig => ({ + accessTeamDomain: "team.cloudflareaccess.com", + accessAud: "aud-tag", + accessNameClaim: "name", + accessGroupsClaim: "groups", + adminEmails: ["admin@example.com"], + organizationId: "default", + organizationName: "Default", + organizationSlug: "default", + secretKey: "x".repeat(32), + allowLocalNetwork: false, + webBaseUrl: "https://localhost", + enableDevAuth: true, + ...overrides, +}); + +describe("cloudflareAccountProvider.listMembers", () => { + it.effect("reports the Access principal so the console can see ADMIN_EMAILS", () => + Effect.gen(function* () { + const provider = yield* AccountProvider; + const { members } = yield* provider.listMembers({}); + expect(members).toEqual([ + { + id: "dev", + userId: "dev", + email: "admin@example.com", + name: "Dev", + avatarUrl: null, + role: "admin", + status: "active", + lastActiveAt: null, + isCurrentUser: true, + }, + ]); + }).pipe(Effect.provide(cloudflareAccountProvider(config()))), + ); + + it.effect("refuses when Access did not authenticate the request", () => + Effect.gen(function* () { + const provider = yield* AccountProvider; + const error = yield* provider.listMembers({}).pipe(Effect.flip); + expect(error).toBeInstanceOf(AccountUnauthorized); + }).pipe(Effect.provide(cloudflareAccountProvider(config({ enableDevAuth: false })))), + ); +}); diff --git a/apps/host-cloudflare/src/account/account-provider.ts b/apps/host-cloudflare/src/account/account-provider.ts index bbbbd3d522..b6260689aa 100644 --- a/apps/host-cloudflare/src/account/account-provider.ts +++ b/apps/host-cloudflare/src/account/account-provider.ts @@ -17,10 +17,10 @@ import type { CloudflareConfig } from "../config"; // uses), reading the `Cf-Access-Jwt-Assertion` header off the request. // // Single-tenant + Access-managed: members, roles, and API keys live in -// Cloudflare Access, NOT in the app. The shell hides the API-keys footer and -// shows no members page, so those methods are never reached from the UI; they -// return empty (reads) or a clear "managed by Cloudflare Access" error (writes) -// to satisfy the provider shape. +// Cloudflare Access, NOT in the app. Writes stay refused. `listMembers` still +// has to return the current Access principal — the console infers admin from +// that list (`isCurrentUser` + role), and an empty list fail-closes every +// workspace-admin action even when `ADMIN_EMAILS` granted `orgRole: "admin"`. // --------------------------------------------------------------------------- const NOT_IN_APP = "Managed by Cloudflare Access, not in the app."; @@ -66,7 +66,28 @@ export const cloudflareAccountProvider = ( listOrgApiKeys: () => Effect.succeed({ apiKeys: [] }), createOrgApiKey: () => forbiddenWrite, revokeOrgApiKey: () => forbiddenWrite, - listMembers: () => Effect.succeed({ members: [] }), + listMembers: (headers) => + principalFrom(headers).pipe( + Effect.flatMap((principal) => + principal + ? Effect.succeed({ + members: [ + { + id: principal.accountId, + userId: principal.accountId, + email: principal.email.length > 0 ? principal.email : null, + name: principal.name, + avatarUrl: principal.avatarUrl, + role: principal.orgRole === "admin" ? "admin" : "member", + status: "active", + lastActiveAt: null, + isCurrentUser: true, + }, + ], + }) + : Effect.fail(new AccountUnauthorized()), + ), + ), listRoles: () => Effect.succeed({ roles: [] }), inviteMember: () => forbiddenWrite, removeMember: () => forbiddenWrite, From 541d6f048bf620fc3fd4bbe380050acaaa868419 Mon Sep 17 00:00:00 2001 From: Ian Pascoe Date: Fri, 25 Sep 2026 05:46:48 -0400 Subject: [PATCH 2/4] 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 6812b29ca7dafcf040d491c67e99498e4fe7b548 Mon Sep 17 00:00:00 2001 From: Ian Pascoe Date: Fri, 25 Sep 2026 05:50:48 -0400 Subject: [PATCH 3/4] 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 }), ); }); From 82fde59d108c2c4b375d6a1dfc8c12e8919910a6 Mon Sep 17 00:00:00 2001 From: Ian Pascoe Date: Fri, 25 Sep 2026 06:01:59 -0400 Subject: [PATCH 4/4] Rebuild tool catalogs by upsert-then-prune so reads never see an emptied catalog --- .changeset/gap-free-catalog-rebuild.md | 5 + packages/core/sdk/src/catalog-persist.test.ts | 383 ++++++++++++++++++ packages/core/sdk/src/executor.ts | 97 ++++- packages/core/sdk/src/shape-memory.ts | 5 +- 4 files changed, 477 insertions(+), 13 deletions(-) create mode 100644 .changeset/gap-free-catalog-rebuild.md create mode 100644 packages/core/sdk/src/catalog-persist.test.ts diff --git a/.changeset/gap-free-catalog-rebuild.md b/.changeset/gap-free-catalog-rebuild.md new file mode 100644 index 0000000000..241fee29f2 --- /dev/null +++ b/.changeset/gap-free-catalog-rebuild.md @@ -0,0 +1,5 @@ +--- +"@executor-js/sdk": patch +--- + +Tool-catalog rebuilds no longer empty the catalog while they run. A rebuild now upserts the new tool and definition rows and then prunes only the names the upstream stopped listing, instead of deleting every row and re-inserting. On databases without interactive transactions (Cloudflare D1), each statement commits on its own, so a search during a rebuild used to find zero tools for that connection, and a rebuild cut off partway (or overlapping another session's rebuild) left the catalog partial; both now keep a complete catalog, and an interrupted rebuild stays stale and retries. Catalog rows are written in size-bounded calls, so a large spec's catalog (Cloudflare's own API) no longer exceeds D1's 32MiB batch limit. diff --git a/packages/core/sdk/src/catalog-persist.test.ts b/packages/core/sdk/src/catalog-persist.test.ts new file mode 100644 index 0000000000..b11146c5fb --- /dev/null +++ b/packages/core/sdk/src/catalog-persist.test.ts @@ -0,0 +1,383 @@ +import { describe, expect, it } from "@effect/vitest"; +import { Effect, Fiber } from "effect"; +import { withQueryContext } from "@executor-js/fumadb/query"; + +import { collectTables, createExecutor } from "./executor"; +import { createExecutorFumaDb } from "./executor-fuma-db"; +import { StorageError, type FumaDb } from "./fuma-runtime"; +import { + AuthTemplateSlug, + ConnectionName, + IntegrationSlug, + ProviderItemId, + ProviderKey, + ToolName, +} from "./ids"; +import { definePlugin } from "./plugin"; +import type { CredentialProvider } from "./provider"; +import { createSqliteTestFumaDb } from "./sqlite-test-db"; +import { makeTestConfig } from "./testing"; + +// --------------------------------------------------------------------------- +// Catalog rebuilds on an auto-commit adapter. +// +// Cloudflare D1 has no interactive transactions (`interactiveTransactions: +// false`), so every statement of a catalog rebuild commits on its own and is +// visible to concurrent reads, and a rebuild cut off partway leaves exactly +// the statements that ran. These cases run the rebuild on that adapter shape +// and pin what a reader and a failed rebuild may observe. +// --------------------------------------------------------------------------- + +const INTEG = IntegrationSlug.make("catalog"); +const TEMPLATE = AuthTemplateSlug.make("apiKey"); +const MAIN = ConnectionName.make("main"); + +interface CatalogTool { + readonly name: string; + readonly description: string; +} + +const memoryProvider = (): CredentialProvider => { + const store = new Map(); + return { + key: ProviderKey.make("memory"), + writable: true, + get: (id) => Effect.sync(() => store.get(String(id)) ?? null), + set: (id, value) => Effect.sync(() => void store.set(String(id), value)), + has: (id) => Effect.sync(() => store.has(String(id))), + list: () => + Effect.sync(() => + Array.from(store.keys()).map((key) => ({ id: ProviderItemId.make(key), name: key })), + ), + }; +}; + +// D1 runs a multi-statement bulk write as one native batch RPC and rejects it +// past 32MiB. The harness adapter enforces the same kind of cap, scaled down +// so a test catalog can cross it. +const BULK_WRITE_CAP_BYTES = 2 * 1024 * 1024; + +// Hooks the tests arm on the adapter handle the executor writes through. +interface PersistHooks { + // Pause right after the rebuild's first write to `tool` commits. + pause: { + readonly reached: PromiseWithResolvers; + readonly release: PromiseWithResolvers; + } | null; + // Reject every write of new `tool` rows (inserts and upserts; deletes + // still run), as when a rebuild is cut off before its rows land. + failToolInserts: boolean; + // Tables each `deleteMany` targeted. + readonly deletes: string[]; + // Resolved when a connection's `tools_synced_at` is stamped. + stamped: PromiseWithResolvers | null; +} + +const hookCatalogWrites = (db: FumaDb, hooks: PersistHooks): FumaDb => { + const toolWrite = async ( + table: string, + write: () => Promise, + kind: "insert" | "delete", + rows: readonly unknown[] = [], + ): Promise => { + if (JSON.stringify(rows).length > BULK_WRITE_CAP_BYTES) { + // oxlint-disable-next-line executor/no-promise-reject -- boundary: fault-injecting FumaDB adapter must reject to emulate D1's batch payload cap + return Promise.reject( + new StorageError({ + message: "Bulk write exceeds the batch payload cap.", + cause: undefined, + }), + ); + } + if (table === "tool" && kind === "insert" && hooks.failToolInserts) { + // oxlint-disable-next-line executor/no-promise-reject -- boundary: fault-injecting FumaDB adapter must reject to exercise a failed rebuild + return Promise.reject( + new StorageError({ message: "Injected tool write failure.", cause: undefined }), + ); + } + const result = await write(); + const pause = table === "tool" ? hooks.pause : null; + if (pause) { + hooks.pause = null; + pause.reached.resolve(); + await pause.release.promise; + } + return result; + }; + // The proxy target is an empty stand-in: the ORM handle's own `withContext` + // is a non-configurable property, which a Proxy over it may not replace. + const wrap = (source: FumaDb): FumaDb => + new Proxy({} as FumaDb, { + get(_target, property) { + if (property === "withContext") { + const withContext = source.withContext?.bind(source); + return withContext === undefined + ? undefined + : (context: unknown) => wrap(withContext(context)); + } + if (property === "transaction") { + const transaction: FumaDb["transaction"] = (run) => + source.transaction((transactionDb) => run(wrap(transactionDb))); + return transaction; + } + if (property === "createMany") { + const createMany: FumaDb["createMany"] = (table, rows) => + toolWrite(table, () => source.createMany(table, rows), "insert", rows); + return createMany; + } + if (property === "upsertMany") { + const upsertMany: FumaDb["upsertMany"] = (table, options) => + toolWrite(table, () => source.upsertMany(table, options), "insert", options.values); + return upsertMany; + } + if (property === "deleteMany") { + const deleteMany: FumaDb["deleteMany"] = (table, options) => { + hooks.deletes.push(table); + return toolWrite(table, () => source.deleteMany(table, options), "delete"); + }; + return deleteMany; + } + if (property === "updateMany") { + const updateMany: FumaDb["updateMany"] = async (table, options) => { + await source.updateMany(table, options); + if (table === "connection" && options.set.tools_synced_at != null) { + hooks.stamped?.resolve(); + } + }; + return updateMany; + } + const value: unknown = Reflect.get(source, property); + return typeof value === "function" ? value.bind(source) : value; + }, + }); + return wrap(db); +}; + +const makeCatalogExecutor = (options: { readonly toolsSyncGraceMs: number | null }) => + Effect.gen(function* () { + const catalog: { tools: readonly CatalogTool[]; definitions: Record } = { + tools: [], + definitions: {}, + }; + const plugin = definePlugin(() => ({ + id: "catalog" as const, + credentialProviders: [memoryProvider()], + storage: () => ({}), + resolveTools: () => + Effect.sync(() => ({ + tools: catalog.tools.map((tool) => ({ + name: ToolName.make(tool.name), + description: tool.description, + })), + definitions: catalog.definitions, + })), + invokeTool: ({ toolRow }) => Effect.succeed({ ran: toolRow.name }), + extension: (ctx) => ({ + seed: () => + ctx.core.integrations.register({ slug: INTEG, description: "Catalog", config: {} }), + }), + }))(); + + const config = makeTestConfig({ plugins: [plugin] as const }); + const sqlite = yield* Effect.acquireRelease( + Effect.promise(() => createSqliteTestFumaDb({ tables: collectTables() })), + (handle) => Effect.promise(() => handle.close()), + ); + // The D1 host's adapter options (apps/host-cloudflare/src/db/d1.ts). + const d1 = createExecutorFumaDb(sqlite.drizzle, { + tables: collectTables(), + namespace: "executor_test", + version: "1.0.0", + provider: "sqlite", + interactiveTransactions: false, + maxBoundParameters: 100, + }); + const hooks: PersistHooks = { pause: null, failToolInserts: false, deletes: [], stamped: null }; + // Hooks sit under the query context: the context wrapper's own + // `withContext` is non-configurable, so it cannot be proxied itself. + const db = withQueryContext(hookCatalogWrites(d1.db as FumaDb, hooks), { + tenant: "test-tenant", + subject: "test-subject", + }) as FumaDb; + const executor = yield* Effect.acquireRelease( + createExecutor({ + ...config, + db, + toolsSyncGraceMs: options.toolsSyncGraceMs, + }), + (instance) => instance.close().pipe(Effect.ignore), + ); + + const markStale = Effect.promise(() => + db.updateMany("connection", { + where: (b) => b.and(b("integration", "=", String(INTEG)), b("name", "=", String(MAIN))), + set: { tools_synced_at: null }, + }), + ); + const syncedAt = Effect.promise(() => + db.findFirst("connection", { + where: (b) => b.and(b("integration", "=", String(INTEG)), b("name", "=", String(MAIN))), + }), + ).pipe(Effect.map((row) => row?.tools_synced_at ?? null)); + const listTools = executor.tools + .list({ integration: INTEG }) + .pipe( + Effect.map((tools) => + tools.map((tool) => ({ name: String(tool.name), description: tool.description })), + ), + ); + const definitions = Effect.promise(() => + db.findMany("definition", { where: (b) => b("integration", "=", String(INTEG)) }), + ).pipe( + Effect.map((rows) => + Object.fromEntries(rows.map((row) => [String(row.name), row.schema] as const)), + ), + ); + + const seed = (tools: readonly CatalogTool[], defs: Record) => + Effect.gen(function* () { + catalog.tools = tools; + catalog.definitions = defs; + yield* executor.catalog.seed(); + yield* executor.connections.create({ + owner: "org", + name: MAIN, + integration: INTEG, + template: TEMPLATE, + value: "secret-token", + }); + }); + + return { catalog, hooks, seed, markStale, syncedAt, listTools, definitions }; + }); + +const byName = (tools: readonly CatalogTool[]) => + [...tools].sort((left, right) => left.name.localeCompare(right.name)); + +describe("catalog rebuild on an auto-commit adapter", () => { + it.effect("a read during a rebuild sees the full catalog, and the rebuild lands exactly", () => + Effect.scoped( + Effect.gen(function* () { + // No grace budget: reads answer from the persisted rows immediately + // while the rebuild runs detached, which is what lets a read land in + // the middle of it. + const harness = yield* makeCatalogExecutor({ toolsSyncGraceMs: 0 }); + yield* harness.seed( + [ + { name: "deploy", description: "deploy v1" }, + { name: "list", description: "list" }, + { name: "legacy", description: "legacy" }, + ], + { Shared: { type: "string" }, LegacyOnly: { type: "number" } }, + ); + expect(byName(yield* harness.listTools).map((tool) => tool.name)).toEqual([ + "deploy", + "legacy", + "list", + ]); + + // The upstream now drops `legacy`, adds `fresh`, and revises `deploy` + // and the `Shared` definition. + harness.catalog.tools = [ + { name: "deploy", description: "deploy v2" }, + { name: "list", description: "list" }, + { name: "fresh", description: "fresh" }, + ]; + harness.catalog.definitions = { Shared: { type: "boolean" }, FreshOnly: { type: "null" } }; + const pause = { + reached: Promise.withResolvers(), + release: Promise.withResolvers(), + }; + harness.hooks.pause = pause; + harness.hooks.stamped = Promise.withResolvers(); + yield* harness.markStale; + + const trigger = yield* Effect.forkChild(harness.listTools); + yield* Effect.promise(() => pause.reached.promise); + + // Mid-rebuild: every tool of the previous catalog is still listed. + const during = (yield* harness.listTools).map((tool) => tool.name); + expect(during).toEqual(expect.arrayContaining(["deploy", "legacy", "list"])); + + pause.release.resolve(); + yield* Fiber.join(trigger); + yield* Effect.promise(() => harness.hooks.stamped!.promise); + + expect(byName(yield* harness.listTools)).toEqual([ + { name: "deploy", description: "deploy v2" }, + { name: "fresh", description: "fresh" }, + { name: "list", description: "list" }, + ]); + expect(yield* harness.definitions).toEqual({ + Shared: { type: "boolean" }, + FreshOnly: { type: "null" }, + }); + expect(yield* harness.syncedAt).not.toBeNull(); + }), + ), + ); + + it.effect("rebuilding an unchanged catalog deletes nothing", () => + Effect.scoped( + Effect.gen(function* () { + const harness = yield* makeCatalogExecutor({ toolsSyncGraceMs: null }); + const tools = [ + { name: "deploy", description: "deploy" }, + { name: "list", description: "list" }, + ]; + yield* harness.seed(tools, { Shared: { type: "string" } }); + + harness.hooks.deletes.length = 0; + yield* harness.markStale; + expect(byName(yield* harness.listTools)).toEqual(tools); + + expect(harness.hooks.deletes).toEqual([]); + expect(yield* harness.syncedAt).not.toBeNull(); + }), + ), + ); + + it.effect("a rebuild that fails partway keeps every row and stays stale", () => + Effect.scoped( + Effect.gen(function* () { + const harness = yield* makeCatalogExecutor({ toolsSyncGraceMs: null }); + const tools = [ + { name: "deploy", description: "deploy" }, + { name: "list", description: "list" }, + ]; + yield* harness.seed(tools, { Shared: { type: "string" } }); + + harness.catalog.tools = [{ name: "fresh", description: "fresh" }]; + harness.catalog.definitions = { FreshOnly: { type: "null" } }; + harness.hooks.failToolInserts = true; + yield* harness.markStale; + + // The failed rebuild is logged and swallowed; the read answers from + // whatever rows the rebuild left behind. + expect(byName(yield* harness.listTools)).toEqual(tools); + expect(yield* harness.syncedAt).toBeNull(); + }), + ), + ); + + it.effect("a catalog larger than one bulk-write payload still lands in full", () => + Effect.scoped( + Effect.gen(function* () { + const harness = yield* makeCatalogExecutor({ toolsSyncGraceMs: null }); + yield* harness.seed([{ name: "deploy", description: "deploy" }], {}); + + // ~3MB of tool rows: past the adapter's per-call payload cap, the way + // Cloudflare's own API catalog is past D1's. + harness.catalog.tools = Array.from({ length: 400 }, (_, index) => ({ + name: `op_${String(index).padStart(3, "0")}`, + description: "x".repeat(8 * 1024), + })); + yield* harness.markStale; + + const tools = yield* harness.listTools; + expect(tools).toHaveLength(400); + expect(yield* harness.syncedAt).not.toBeNull(); + }), + ), + ); +}); diff --git a/packages/core/sdk/src/executor.ts b/packages/core/sdk/src/executor.ts index 6343dd39ad..cd46f9f053 100644 --- a/packages/core/sdk/src/executor.ts +++ b/packages/core/sdk/src/executor.ts @@ -227,7 +227,19 @@ import { annotateToolResultOutcome, isToolResult } from "./tool-result"; import { makeShapeMemory, observedShapeToJsonSchema, SHAPE_MEMORY_PLUGIN_ID } from "./shape-memory"; import { isUnauthorizedToolFailure } from "./auth-tool-failure"; -const PLUGIN_STORAGE_DELETE_KEY_BATCH_SIZE = 90; +// Values per `in (...)` predicate on batched deletes: keeps each statement +// under D1's 100-bound-parameter limit alongside its scope columns. +const DELETE_IN_BATCH_SIZE = 90; + +// Unique key shared by the `tool` and `definition` catalog tables. +const CATALOG_ROW_KEY = ["tenant", "owner", "subject", "integration", "connection", "name"]; + +// Serialized row bytes per catalog upsert call. On D1 one multi-statement +// upsert runs as a single native batch RPC, which D1 caps at 32MiB; a large +// spec's full catalog (Cloudflare's own API: ~37MB) exceeds that in one call, +// and building it stalls the isolate for seconds. The budget leaves room for +// the SQL and RPC encoding the rows expand into. +const CATALOG_UPSERT_CHUNK_BYTES = 1024 * 1024; const MAX_APPROVAL_ARGUMENT_PREVIEW_CHARS = 4_000; // --------------------------------------------------------------------------- @@ -1673,12 +1685,8 @@ const makePluginStorageFacade = (input: { Effect.gen(function* () { for (const [collection, keys] of keysByCollection(entries)) { const uniqueKeys = [...keys]; - for ( - let offset = 0; - offset < uniqueKeys.length; - offset += PLUGIN_STORAGE_DELETE_KEY_BATCH_SIZE - ) { - const batchKeys = uniqueKeys.slice(offset, offset + PLUGIN_STORAGE_DELETE_KEY_BATCH_SIZE); + for (let offset = 0; offset < uniqueKeys.length; offset += DELETE_IN_BATCH_SIZE) { + const batchKeys = uniqueKeys.slice(offset, offset + DELETE_IN_BATCH_SIZE); yield* input.core.deleteMany("plugin_storage", { where: (b) => b.and( @@ -3543,6 +3551,53 @@ export const createExecutor = (effect: Effect.Effect) => catalogPersistLock.withPermits(1)(transaction(effect)); + // Upsert catalog rows in calls bounded by their serialized size (see + // CATALOG_UPSERT_CHUNK_BYTES). Chunks commit independently, which the + // upsert-then-prune rebuild tolerates: a cut-off rebuild leaves every row + // intact and the connection unstamped. + const upsertCatalogRows = ( + table: "tool" | "definition", + update: readonly string[], + rows: readonly Record[], + ) => + Effect.gen(function* () { + let chunk: Record[] = []; + let chunkBytes = 0; + for (const row of rows) { + const rowBytes = JSON.stringify(row).length; + if (chunk.length > 0 && chunkBytes + rowBytes > CATALOG_UPSERT_CHUNK_BYTES) { + yield* core.upsertMany(table, { target: CATALOG_ROW_KEY, update, values: chunk }); + chunk = []; + chunkBytes = 0; + } + chunk.push(row); + chunkBytes += rowBytes; + } + yield* core.upsertMany(table, { target: CATALOG_ROW_KEY, update, values: chunk }); + }); + + // Delete the connection's catalog rows whose names the new listing no + // longer produces. Reads names only, so an unchanged catalog costs one + // narrow read and no writes. + const pruneCatalogRows = ( + table: "tool" | "definition", + where: CoreWhere, + kept: readonly { readonly name: string }[], + ) => + Effect.gen(function* () { + const keptNames = new Set(kept.map((row) => row.name)); + const existing = yield* core.findMany(table, { where, select: ["name"] }); + const staleNames = existing + .map((row) => String(row.name)) + .filter((name) => !keptNames.has(name)); + for (let offset = 0; offset < staleNames.length; offset += DELETE_IN_BATCH_SIZE) { + const batch = staleNames.slice(offset, offset + DELETE_IN_BATCH_SIZE); + yield* core.deleteMany(table, { + where: (b: AnyCb) => b.and(where(b), b("name", "in", batch)), + }); + } + }); + const produceConnectionToolsUnshared = ( integrationRow: IntegrationRow, ref: ConnectionRef, @@ -3766,12 +3821,32 @@ export const createExecutor =