diff --git a/.changeset/tools-read-stale-sync-grace.md b/.changeset/tools-read-stale-sync-grace.md new file mode 100644 index 0000000000..cb570bd4ac --- /dev/null +++ b/.changeset/tools-read-stale-sync-grace.md @@ -0,0 +1,11 @@ +--- +"executor": patch +--- + +**Tools reads stop waiting on slow upstream servers** + +A tools read rebuilds every connection whose catalog has gone stale before answering. The rebuilds already ran concurrently, but the read still waited for all of them, so one slow or unreachable MCP server gated every catalog read behind its network timeout — a tools listing could take tens of seconds while healthy connections sat ready. + +A read now waits at most a short grace budget (2 seconds by default) for the rebuilds, then answers from the persisted catalog. The rebuilds keep running after the read returns and land on a later read, so the catalog still converges — it just no longer holds the reader hostage while it does. Overlapping reads share one in-flight rebuild per connection instead of stacking new ones. + +The budget is `toolsSyncGraceMs` on the SDK config. Pass `null` to restore the strict behavior, where a read blocks until every rebuild finishes and always reflects a fully converged catalog. diff --git a/apps/cloud/src/engine/execution-stack.ts b/apps/cloud/src/engine/execution-stack.ts index a7ff4de700..230f028fdf 100644 --- a/apps/cloud/src/engine/execution-stack.ts +++ b/apps/cloud/src/engine/execution-stack.ts @@ -32,7 +32,7 @@ // seams module; the decorator is composed on top. // --------------------------------------------------------------------------- -import { env } from "cloudflare:workers"; +import { env, waitUntil } from "cloudflare:workers"; import { Layer } from "effect"; import { @@ -199,6 +199,10 @@ export const CloudHostConfig: Layer.Layer = Layer.sync(HostConfig, ( // user-selectable provider surface. exposeCredentialProviders: false, firstPartyOAuthClients: cloudFirstPartyOAuthClients(), + // Workers cancel request-scoped I/O once the response settles; the ambient + // `waitUntil` binds to the in-flight invocation (HTTP request or DO call), + // so stale tool-catalog rebuilds that outlive a read still converge. + waitUntil, // Enterprise-managed authorization ships behind a PostHog flag. Cloud is the // one host with a flag service, so cloud is the one host that installs a // gate; everywhere else the seam stays empty and the profile is attempted as diff --git a/packages/core/api/src/server/scoped-executor.ts b/packages/core/api/src/server/scoped-executor.ts index af1c35244d..426363c6b2 100644 --- a/packages/core/api/src/server/scoped-executor.ts +++ b/packages/core/api/src/server/scoped-executor.ts @@ -123,6 +123,15 @@ export interface HostConfigShape { * operator knob. */ readonly toolsSyncTtlMs?: number | null; + /** + * Forwarded verbatim to `ExecutorConfig.waitUntil`: the host's keep-alive + * for background work that outlives a request (stale tool-catalog rebuilds + * that keep running after a read stops waiting). Cloud supplies the + * platform `waitUntil` from `cloudflare:workers`, which binds to the + * in-flight invocation ambiently; long-lived hosts (self-host, local, + * tests) omit it and detached fibers simply run to completion in-process. + */ + readonly waitUntil?: (promise: Promise) => void; } export class HostConfig extends Context.Service()( @@ -305,6 +314,7 @@ export const makeScopedExecutor = < fetch: hostedFetch, onIntegrationChange: config.onIntegrationChange, ...(config.toolsSyncTtlMs !== undefined ? { toolsSyncTtlMs: config.toolsSyncTtlMs } : {}), + ...(config.waitUntil !== undefined ? { waitUntil: config.waitUntil } : {}), onElicitation: "accept-all", redirectUri, oauthCallbackStateOrgSlug: orgSlug, diff --git a/packages/core/sdk/src/executor.ts b/packages/core/sdk/src/executor.ts index b9a350cdc2..7c3c25929d 100644 --- a/packages/core/sdk/src/executor.ts +++ b/packages/core/sdk/src/executor.ts @@ -2,6 +2,7 @@ import { Deferred, Duration, Effect, + Fiber, Inspectable, Layer, Option, @@ -694,6 +695,23 @@ export interface ExecutorConfig) => void; /** * Notified after a durable integration-catalog change commits (a row * created or removed). Best-effort observation only: the notification runs @@ -727,6 +745,13 @@ export interface ExecutorConfig + Effect.gen(function* () { + const fiber = yield* Effect.forkDetach( + syncStaleConnectionTools.pipe( + Effect.catch((error) => + Effect.logWarning("executor stale tool sync scan failed", { + error: describeSyncFailure(error), + }), + ), + ), + ); + // On hosts that cancel request-scoped I/O once the response settles + // (Cloudflare Workers), hand the host the rebuilds' completion so the + // catalog still converges after the read stops waiting. + config.waitUntil?.( + new Promise((resolve) => fiber.addObserver(() => resolve(undefined))), + ); + yield* Fiber.await(fiber).pipe(Effect.timeoutOption(graceMs), Effect.asVoid); + }); + const toolsList = (filter?: ToolListFilter): Effect.Effect => Effect.gen(function* () { - yield* syncStaleConnectionTools; + if (toolsSyncGraceMs === null) { + yield* syncStaleConnectionTools; + } else { + yield* awaitStaleSyncWithinGrace(toolsSyncGraceMs); + } // Projected: the list surface is metadata (address, description, // annotations) — loading every tool's input/output schema JSON made // an unbounded list scale with schema bytes, not tool count. diff --git a/packages/core/sdk/src/index.ts b/packages/core/sdk/src/index.ts index a1e83c4420..524459517e 100644 --- a/packages/core/sdk/src/index.ts +++ b/packages/core/sdk/src/index.ts @@ -424,6 +424,7 @@ export { type ExecutorDbFactory, type ExecutorDbInput, type ParsedToolAddress, + DEFAULT_TOOLS_SYNC_GRACE_MS, STALE_TOOLS_SYNC_CONCURRENCY, createExecutor, collectTables, diff --git a/packages/plugins/mcp/src/sdk/catalog-sync.test.ts b/packages/plugins/mcp/src/sdk/catalog-sync.test.ts index 03dd2a6e45..f3328876b3 100644 --- a/packages/plugins/mcp/src/sdk/catalog-sync.test.ts +++ b/packages/plugins/mcp/src/sdk/catalog-sync.test.ts @@ -42,11 +42,17 @@ const TEMPLATE = AuthTemplateSlug.make("none"); const makeCatalogTestExecutor = ( serverUrl: string, - options?: { readonly toolsSyncTtlMs?: number | null }, + options?: { + readonly toolsSyncTtlMs?: number | null; + readonly toolsSyncGraceMs?: number | null; + }, ) => createExecutor({ ...makeTestConfig({ plugins: [memoryCredentialsPlugin(), mcpPlugin()] as const }), ...(options?.toolsSyncTtlMs === undefined ? {} : { toolsSyncTtlMs: options.toolsSyncTtlMs }), + ...(options?.toolsSyncGraceMs === undefined + ? {} + : { toolsSyncGraceMs: options.toolsSyncGraceMs }), }).pipe( Effect.tap((executor) => Effect.gen(function* () { @@ -355,6 +361,11 @@ describe("MCP stale-catalog refresh", () => { // 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++) { @@ -401,3 +412,95 @@ describe("MCP stale-catalog refresh", () => { }), ); }); + +// --------------------------------------------------------------------------- +// Stale-refresh grace budget. +// +// A tools read waits at most `toolsSyncGraceMs` for stale rebuilds, then +// answers from the persisted catalog while the rebuilds finish detached. One +// slow upstream server must bound neither the read nor convergence: the read +// serves the stale-but-working rows now, and a later read reflects the +// re-listed catalog once the server finally answers. +// --------------------------------------------------------------------------- + +const serveLatchedMutableServer = () => + Effect.gen(function* () { + const catalog = yield* Ref.make("alpha"); + const armed = yield* Ref.make(false); + const release = yield* Deferred.make(); + + const server = yield* serveTestHttpApp((request) => + Effect.gen(function* () { + if (request.method === "GET") { + return HttpServerResponse.text("SSE disabled", { status: 405 }); + } + const body = yield* request.text.pipe(Effect.orDie); + const rpc = Option.getOrUndefined(decodeJsonRpcRequest(body)); + if (!rpc) { + return HttpServerResponse.text("Invalid JSON-RPC fixture request", { status: 400 }); + } + if (rpc.method === "initialize") { + return jsonRpcResult(rpc, { + protocolVersion: "2025-06-18", + capabilities: { tools: { listChanged: true } }, + serverInfo: { name: "latched-mutable-fixture", version: "1.0.0" }, + }); + } + if (rpc.method === "notifications/initialized") { + return HttpServerResponse.text("", { status: 202 }); + } + if (rpc.method !== "tools/list") { + return HttpServerResponse.text("Unexpected JSON-RPC method", { status: 400 }); + } + // Once armed, park every listing until released — the "slow server". + if (yield* Ref.get(armed)) { + yield* Deferred.await(release); + } + return jsonRpcResult(rpc, { tools: [pageTool(yield* Ref.get(catalog))] }); + }), + ); + + return { + url: server.url("/mcp"), + rename: Ref.set(catalog, "beta"), + arm: Ref.set(armed, true), + release: Deferred.succeed(release, undefined), + } as const; + }); + +describe("MCP stale-refresh grace budget", () => { + // `it.live` (real clock): the grace timeout must actually fire while a real + // HTTP listing stays parked. + it.live("a read outlasting the grace serves the stored catalog, then converges", () => + Effect.gen(function* () { + const fixture = yield* serveLatchedMutableServer(); + const executor = yield* makeCatalogTestExecutor(fixture.url, { + // Every read finds the catalog expired, and waits at most 100ms. + toolsSyncTtlMs: 0, + toolsSyncGraceMs: 100, + }); + + expect(toolNames(yield* executor.tools.list())).toContain("alpha"); + + // The server's catalog changes AND the server stops answering listings. + yield* fixture.rename; + yield* fixture.arm; + + // The re-list is parked, so only the grace path can produce an answer — + // and it is the persisted (stale) catalog, not a failure or a hang. + expect(toolNames(yield* executor.tools.list())).toContain("alpha"); + + // Once the server answers, the detached rebuild lands and a later read + // reflects the re-listed catalog. + yield* fixture.release; + const converged = yield* Effect.gen(function* () { + while (true) { + const names = toolNames(yield* executor.tools.list()); + if (names.includes("beta")) return names; + yield* Effect.sleep("100 millis"); + } + }).pipe(Effect.timeoutOption("10 seconds")); + expect(Option.isSome(converged)).toBe(true); + }), + ); +});