diff --git a/.changeset/concurrent-stale-tools-sync.md b/.changeset/concurrent-stale-tools-sync.md new file mode 100644 index 0000000000..8f4bc858ca --- /dev/null +++ b/.changeset/concurrent-stale-tools-sync.md @@ -0,0 +1,11 @@ +--- +"executor": patch +--- + +**Stale tool catalogs refresh together instead of one after another, and self-host can set the freshness window** + +A tools read rebuilds every connection whose catalog has gone stale. Those rebuilds each dial their own upstream, but ran strictly one after another, so a host with several stale remote catalogs paid the sum of every server's latency on the read that tripped the TTL. The upstream listings now run concurrently, bounded so a large stale set cannot open an unbounded number of listings from one read. + +Only the listings overlap. Each rebuild's catalog write stays single-file, because a self-host database is one connection issuing raw `BEGIN`/`COMMIT` and a second transaction opened while one is live fails outright. A rebuild that fails now also logs a warning naming the connection and the reason, instead of disappearing: the read still succeeds on the stale-but-working catalog and the other connections still finish, but a permanently broken connection no longer re-fails silently on every read. + +Self-host also exposes the freshness window as `EXECUTOR_TOOLS_SYNC_TTL_MS`. Leave it unset for the 15-minute default, or set `off` (equivalently `null` or `false`, in any case) to disable time-based re-sync and leave stale-marking and config revision as the only refresh triggers. The value forwards to the SDK verbatim, so `0` keeps its SDK meaning: every catalog is expired on every read. A malformed, negative, or too-large-to-represent value is refused at boot rather than silently falling back to the default. diff --git a/apps/host-selfhost/src/config.ts b/apps/host-selfhost/src/config.ts index dae13a5e16..acb77204f3 100644 --- a/apps/host-selfhost/src/config.ts +++ b/apps/host-selfhost/src/config.ts @@ -51,6 +51,12 @@ export interface SelfHostConfig { * minutes (the same pattern as MCP_PAUSED_SESSION_IDLE_TIMEOUT_MS on cloud). */ readonly sandboxTimeoutMs: number | undefined; + /** + * How long a connection's persisted remote tool catalog stays fresh, in ms. + * `undefined` takes the SDK default (15 minutes); `null` disables time-based + * re-sync, leaving stale-marking and config revision as the only triggers. + */ + readonly toolsSyncTtlMs: number | null | undefined; } export const resolveDataDir = (): string => @@ -157,6 +163,7 @@ export const loadConfig = (): SelfHostConfig => { organizationName: process.env.EXECUTOR_ORG_NAME ?? "Default", orgSlug: resolveOrgSlug(), sandboxTimeoutMs: resolveSandboxTimeoutMs(), + toolsSyncTtlMs: resolveToolsSyncTtlMs(), }; }; @@ -190,3 +197,42 @@ const resolveOrgSlug = (): string => { } return slug; }; + +// EXECUTOR_TOOLS_SYNC_TTL_MS — how long a remote tool catalog (an MCP server's +// tool set, which changes server-side with no executor-visible signal) stays +// fresh before the next tools read re-lists it. Unset takes the SDK default of +// 15 minutes. +// +// The value forwards to the SDK's `toolsSyncTtlMs` verbatim, so `0` keeps the +// SDK's meaning — every catalog is expired on every read. "off", "null" and +// "false" disable time-based re-sync (the SDK's `null` sentinel), since +// operators reach for all three spellings. The comparison is case-insensitive: +// "OFF" and "False" are the same intent typed by a different operator. +// +// Like the other knobs here a malformed or negative value is refused rather +// than silently ignored: an operator who sets the TTL and typos it should find +// out at boot, not by wondering months later why catalogs never refresh. +const TOOLS_SYNC_TTL_DISABLE_TOKENS = new Set(["off", "null", "false"]); + +const resolveToolsSyncTtlMs = (): number | null | undefined => { + const raw = process.env.EXECUTOR_TOOLS_SYNC_TTL_MS?.trim(); + if (!raw) return undefined; + if (TOOLS_SYNC_TTL_DISABLE_TOKENS.has(raw.toLowerCase())) return null; + const parsed = Number(raw); + // `isSafeInteger`, not `isInteger`: past 2^53 a decimal literal silently + // rounds to a nearby representable value, so an operator's typo'd digit + // would boot as a TTL they never wrote. Refuse it instead. + if (!Number.isSafeInteger(parsed)) { + // oxlint-disable-next-line executor/no-try-catch-or-throw, executor/no-error-constructor -- boundary: refuse to boot on a malformed operator knob + throw new Error( + `EXECUTOR_TOOLS_SYNC_TTL_MS ${JSON.stringify(raw)} is not an exactly representable whole number of milliseconds ("off", "null" or "false" disable time-based re-sync)`, + ); + } + if (parsed < 0) { + // oxlint-disable-next-line executor/no-try-catch-or-throw, executor/no-error-constructor -- boundary: refuse to boot on a malformed operator knob + throw new Error( + `EXECUTOR_TOOLS_SYNC_TTL_MS ${JSON.stringify(raw)} must not be negative (use "off" to disable time-based re-sync)`, + ); + } + return parsed; +}; diff --git a/apps/host-selfhost/src/execution.ts b/apps/host-selfhost/src/execution.ts index aa2ffee536..8dab577b93 100644 --- a/apps/host-selfhost/src/execution.ts +++ b/apps/host-selfhost/src/execution.ts @@ -55,6 +55,7 @@ export const SelfHostHostConfig: Layer.Layer = Layer.sync(HostConfig allowLocalNetwork: config.allowLocalNetwork, webBaseUrl: config.webBaseUrl, oauthCallbackPath: "/api/oauth/callback", + toolsSyncTtlMs: config.toolsSyncTtlMs, onIntegrationChange: (event) => selfHostAnalytics.record( event.kind === "added" ? "integration_added" : "integration_removed", diff --git a/apps/host-selfhost/src/executor-config.test.ts b/apps/host-selfhost/src/executor-config.test.ts index 0d56bc32f2..313d097b85 100644 --- a/apps/host-selfhost/src/executor-config.test.ts +++ b/apps/host-selfhost/src/executor-config.test.ts @@ -1,11 +1,14 @@ import { afterEach, beforeEach, expect, test } from "@effect/vitest"; +import { loadConfig } from "./config"; import executorConfig from "../executor.config"; const ENV_NAME = "EXECUTOR_ALLOW_STDIO_MCP"; const SECRET_ENV_NAME = "EXECUTOR_SECRET_KEY"; +const TTL_ENV_NAME = "EXECUTOR_TOOLS_SYNC_TTL_MS"; const originalValue = process.env[ENV_NAME]; const originalSecret = process.env[SECRET_ENV_NAME]; +const originalTtl = process.env[TTL_ENV_NAME]; beforeEach(() => { process.env[SECRET_ENV_NAME] = originalSecret ?? "executor-config-test-secret"; @@ -22,6 +25,11 @@ afterEach(() => { } else { process.env[SECRET_ENV_NAME] = originalSecret; } + if (originalTtl === undefined) { + delete process.env[TTL_ENV_NAME]; + } else { + process.env[TTL_ENV_NAME] = originalTtl; + } }); const allowStdio = (): boolean => { @@ -57,3 +65,50 @@ test("stdio MCP is enabled when the opt-in is exactly true", () => { process.env[ENV_NAME] = "true"; expect(allowStdio()).toBe(true); }); + +test("an unset tools-sync TTL leaves the SDK default in place", () => { + delete process.env[TTL_ENV_NAME]; + expect(loadConfig().toolsSyncTtlMs).toBeUndefined(); + + process.env[TTL_ENV_NAME] = " "; + expect(loadConfig().toolsSyncTtlMs).toBeUndefined(); +}); + +test("a positive tools-sync TTL is forwarded verbatim", () => { + process.env[TTL_ENV_NAME] = "60000"; + expect(loadConfig().toolsSyncTtlMs).toBe(60000); +}); + +// 0 keeps the SDK's own meaning — every catalog is expired on every read — +// so the env var never means the opposite of the config field it feeds. +test("a zero tools-sync TTL forwards as the SDK's always-stale 0", () => { + process.env[TTL_ENV_NAME] = "0"; + expect(loadConfig().toolsSyncTtlMs).toBe(0); +}); + +// Case-insensitive: the disable tokens are operator intent, not a keyword, and +// "OFF" typed in a systemd unit means what "off" means in a .env file. +test.each(["off", "null", "false", "OFF", "Null", "FALSE", " Off "])( + "the tools-sync TTL is disabled by %s", + (raw) => { + process.env[TTL_ENV_NAME] = raw; + expect(loadConfig().toolsSyncTtlMs).toBeNull(); + }, +); + +// A typo'd knob must not silently degrade into the 15-minute default; the +// operator finds out at boot instead of wondering why catalogs never refresh. +// "9007199254740993" and "1e30" are whole numbers that no longer round-trip +// through a double — accepting them would boot a TTL the operator never wrote. +test.each(["abc", "60_000", "1.5", "1e3ms", "NaN", "Infinity", "9007199254740993", "1e30"])( + "a malformed tools-sync TTL (%s) refuses to boot", + (raw) => { + process.env[TTL_ENV_NAME] = raw; + expect(() => loadConfig()).toThrow(/EXECUTOR_TOOLS_SYNC_TTL_MS/); + }, +); + +test("a negative tools-sync TTL refuses to boot", () => { + process.env[TTL_ENV_NAME] = "-1"; + expect(() => loadConfig()).toThrow(/must not be negative/); +}); diff --git a/packages/core/api/src/server/scoped-executor.ts b/packages/core/api/src/server/scoped-executor.ts index 9e6013d230..af1c35244d 100644 --- a/packages/core/api/src/server/scoped-executor.ts +++ b/packages/core/api/src/server/scoped-executor.ts @@ -115,6 +115,14 @@ export interface HostConfigShape { * attempted as it was before the gate existed. */ readonly enterpriseManagedRollout?: ExecutorConfig["enterpriseManagedRollout"]; + /** + * Forwarded verbatim to `ExecutorConfig.toolsSyncTtlMs`: how long a + * connection's persisted remote tool catalog stays fresh. Omit to take the + * SDK default (15 minutes); `null` disables time-based re-sync. Declared + * here — not per-request — because catalog freshness is a deployment-wide + * operator knob. + */ + readonly toolsSyncTtlMs?: number | null; } export class HostConfig extends Context.Service()( @@ -296,6 +304,7 @@ export const makeScopedExecutor = < httpClientLayer, fetch: hostedFetch, onIntegrationChange: config.onIntegrationChange, + ...(config.toolsSyncTtlMs !== undefined ? { toolsSyncTtlMs: config.toolsSyncTtlMs } : {}), onElicitation: "accept-all", redirectUri, oauthCallbackStateOrgSlug: orgSlug, diff --git a/packages/core/sdk/src/connections.test.ts b/packages/core/sdk/src/connections.test.ts index 62f8bae7f1..30944696cb 100644 --- a/packages/core/sdk/src/connections.test.ts +++ b/packages/core/sdk/src/connections.test.ts @@ -1,5 +1,15 @@ import { describe, expect, it } from "@effect/vitest"; -import { Deferred, Effect, Fiber, Predicate, Result, Schema } from "effect"; +import { + Deferred, + Effect, + Fiber, + Inspectable, + Logger, + Option, + Predicate, + Result, + Schema, +} from "effect"; import { AuthTemplateSlug, @@ -11,6 +21,7 @@ import { ToolName, } from "./ids"; import { createExecutor } from "./executor"; +import { StorageError, type FumaDb } from "./fuma-runtime"; import { HealthCheckResult } from "./health-check"; import { definePlugin } from "./plugin"; import type { CredentialProvider } from "./provider"; @@ -44,6 +55,38 @@ const memoryProvider = (): CredentialProvider => { const INTEG = IntegrationSlug.make("vercel"); const TEMPLATE = AuthTemplateSlug.make("apiKey"); +/** Wrap a test `FumaDb` so every transaction it opens is observable. The + * executor re-binds its own owner context onto the handle it is given, so the + * wrapper must forward `withContext` re-wrapped — otherwise the instrument is + * dropped before any executor query runs. */ +const instrumentTransactions = ( + db: FumaDb, + hooks: { readonly enter: () => void; readonly exit: () => void }, +): FumaDb => { + const wrap = (inner: FumaDb): FumaDb => + new Proxy(inner, { + get(target, prop) { + if (prop === "withContext") { + return (context: unknown) => + wrap((target.withContext as (c: unknown) => FumaDb)(context)); + } + if (prop === "transaction") { + return async (run: Parameters[0]) => { + hooks.enter(); + // oxlint-disable-next-line executor/no-try-catch-or-throw -- boundary: test instrument must unwind on both outcomes + try { + return await target.transaction(run); + } finally { + hooks.exit(); + } + }; + } + return Reflect.get(target, prop); + }, + }); + return wrap(db); +}; + const ConnectionListHealthOutput = Schema.Struct({ connections: Schema.Array(Schema.Struct({ lastHealth: Schema.NullOr(HealthCheckResult) })), }); @@ -803,6 +846,200 @@ describe("tool catalog sync safety", () => { }), ), ); + + // A tools read rebuilds every stale connection it finds, and those rebuilds + // run their upstream listings together. Their catalog WRITES must not: the + // self-host database is a single libSQL connection issuing raw BEGIN/COMMIT, + // where a second transaction opened while one is live fails outright. The + // test observes real transactions through the db handle, so it fails if the + // persist step ever loses its permit. + it.effect("overlaps stale discovery but never overlaps catalog persistence", () => + Effect.scoped( + Effect.gen(function* () { + const STALE_CONNECTIONS = 4; + const CONNECTION_NAMES = ["alpha", "beta", "gamma", "delta"] as const; + + let openTransactions = 0; + let maxOpenTransactions = 0; + let discovering = 0; + let latched = false; + const allDiscovering = yield* Deferred.make(); + + const guardedPlugin = definePlugin(() => ({ + id: "guarded" as const, + credentialProviders: [memoryProvider()], + storage: () => ({}), + remoteToolCatalog: true, + // Once latched, no listing answers until every stale connection is + // discovering. A serial fan-out parks on the first one forever, so + // this also proves discovery still overlaps after the restructure. + resolveTools: ({ connection }) => + Effect.gen(function* () { + if (latched) { + discovering += 1; + if (discovering >= STALE_CONNECTIONS) { + yield* Deferred.succeed(allDiscovering, undefined); + } + yield* Deferred.await(allDiscovering); + } + return { + tools: [ + { name: ToolName.make(`deploy_${String(connection.name)}`), description: "d" }, + ], + }; + }), + invokeTool: ({ toolRow }) => Effect.succeed({ ran: toolRow.name }), + extension: (ctx) => ({ + seed: () => + ctx.core.integrations.register({ + slug: INTEG, + description: "Vercel", + config: {}, + }), + }), + }))(); + + const config = makeTestConfig({ plugins: [guardedPlugin] as const }); + const executor = yield* createExecutor({ + ...config, + db: instrumentTransactions(config.db, { + enter: () => { + openTransactions += 1; + maxOpenTransactions = Math.max(maxOpenTransactions, openTransactions); + }, + exit: () => { + openTransactions -= 1; + }, + }), + }); + yield* executor.guarded.seed(); + for (const name of CONNECTION_NAMES) { + yield* executor.connections.create({ + owner: "org", + name: ConnectionName.make(name), + integration: INTEG, + template: TEMPLATE, + value: "secret-token", + }); + } + + // Mark the whole set stale, then arm the latch so the next read is + // purely the stale-refresh fan-out. + yield* Effect.promise(() => + config.db.updateMany("connection", { + where: (b) => b("integration", "=", String(INTEG)), + set: { tools_synced_at: null }, + }), + ); + latched = true; + + // Well inside the harness timeout: a serial fan-out never releases the + // latch and fails the assertion below instead of the whole runner. + const tools = yield* executor.tools + .list({ integration: INTEG }) + .pipe(Effect.timeoutOption("10 seconds")); + + expect(Option.isSome(tools)).toBe(true); + expect(discovering).toBe(STALE_CONNECTIONS); + // The load-bearing assertion: concurrent discovery, single-file writes. + expect(maxOpenTransactions).toBe(1); + }), + ), + ); + + // Partial failure must stay partial AND stay visible. A rebuild that cannot + // reach its upstream keeps the stale-but-working catalog, lets its peers + // finish, and leaves a warning naming the connection — otherwise a + // permanently broken connection re-fails on every read with no trace. + it.effect("a failed stale rebuild warns and neither fails nor blocks the read", () => + Effect.scoped( + Effect.gen(function* () { + let latched = false; + const guardedPlugin = definePlugin(() => ({ + id: "guarded" as const, + credentialProviders: [memoryProvider()], + storage: () => ({}), + remoteToolCatalog: true, + // The realistic failure shape: a plugin reports a StorageError whose + // `cause` carries the actionable upstream detail, exactly as the MCP + // plugin does when a server cannot be reached. + resolveTools: ({ connection }) => + latched && String(connection.name) === "broken" + ? Effect.fail( + new StorageError({ + message: "upstream listing refused", + // oxlint-disable-next-line executor/no-error-constructor -- boundary: the fixture reproduces a real plugin cause, which is a built-in Error + cause: new Error("connect ECONNREFUSED"), + }), + ) + : Effect.succeed({ + tools: [ + { name: ToolName.make(`deploy_${String(connection.name)}`), description: "d" }, + ], + }), + invokeTool: ({ toolRow }) => Effect.succeed({ ran: toolRow.name }), + extension: (ctx) => ({ + seed: () => + ctx.core.integrations.register({ + slug: INTEG, + description: "Vercel", + config: {}, + }), + }), + }))(); + + const config = makeTestConfig({ plugins: [guardedPlugin] as const }); + const executor = yield* createExecutor(config); + yield* executor.guarded.seed(); + for (const name of ["broken", "healthy"]) { + yield* executor.connections.create({ + owner: "org", + name: ConnectionName.make(name), + integration: INTEG, + template: TEMPLATE, + value: "secret-token", + }); + } + + yield* Effect.promise(() => + config.db.updateMany("connection", { + where: (b) => b("integration", "=", String(INTEG)), + set: { tools_synced_at: null }, + }), + ); + latched = true; + + const warnings: string[] = []; + const capture = Logger.make((options) => { + if (options.logLevel === "Warn") { + warnings.push(Inspectable.toStringUnknown(options.message, 0)); + } + }); + const tools = yield* executor.tools + .list({ integration: INTEG }) + .pipe(Effect.provide(Logger.layer([capture]))); + + // The read succeeds, and the failing connection keeps its previously + // persisted catalog rather than being wiped by a failed listing. + expect(tools.map((tool) => String(tool.name)).sort()).toEqual([ + "deploy_broken", + "deploy_healthy", + ]); + + const failureWarning = warnings.find((line) => + line.includes("executor stale tool sync failed"), + ); + expect(failureWarning).toBeDefined(); + expect(failureWarning).toContain("broken"); + // Both halves: the failure and the cause that names what to fix. A bare + // structural render of the error drops the cause entirely. + expect(failureWarning).toContain("upstream listing refused"); + expect(failureWarning).toContain("connect ECONNREFUSED"); + // The healthy peer is not swept into the failure. + expect(failureWarning).not.toContain("healthy"); + }), + ), + ); }); describe("connections.checkHealth", () => { diff --git a/packages/core/sdk/src/executor.ts b/packages/core/sdk/src/executor.ts index f5ad432e79..51e24ab6a1 100644 --- a/packages/core/sdk/src/executor.ts +++ b/packages/core/sdk/src/executor.ts @@ -1,4 +1,14 @@ -import { Deferred, Duration, Effect, Inspectable, Layer, Option, Predicate, Schema } from "effect"; +import { + Deferred, + Duration, + Effect, + Inspectable, + Layer, + Option, + Predicate, + Schema, + Semaphore, +} from "effect"; import { FetchHttpClient, type HttpClient } from "effect/unstable/http"; import { fumadb } from "@executor-js/fumadb"; import { memoryAdapter } from "@executor-js/fumadb/adapters/memory"; @@ -715,6 +725,12 @@ export interface ExecutorConfig storageFailureFromUnknown(`${hook} failed for plugin ${pluginId}`, cause); +// oxlint-disable executor/no-instanceof-error, executor/no-unknown-error-message -- boundary: render an arbitrary failure into one readable log field +/** One-line rendering of a failed rebuild, for the operator-facing warning. + * A `StorageError` carries the actionable detail in its `cause` (the plugin's + * own failure) while its own message only names the hook, and structural + * stringification drops a `cause` that is an `Error` — so unwrap one level and + * keep both halves. */ +const describeSyncFailure = (error: unknown): string => { + const base = + error instanceof Error && error.message.length > 0 + ? error.message + : Inspectable.toStringUnknown(error, 0); + const cause = (error as { readonly cause?: unknown } | null | undefined)?.cause; + if (cause instanceof Error && cause.message.length > 0) return `${base}: ${cause.message}`; + return base; +}; +// oxlint-enable executor/no-instanceof-error, executor/no-unknown-error-message + const createDefaultMemoryDb = (tables: FumaTables): ExecutorDb => { const version = "1.0.0"; const latestSchema = fumaSchema>({ @@ -2874,6 +2907,25 @@ export const createExecutor = result.incompleteReason ?? "plugin returned an incomplete tool catalog"; + // Tool production has two phases with very different shapes: DISCOVERY (the + // plugin's `resolveTools` — network, slow, independent per connection) and + // PERSISTENCE (a short catalog-replacement transaction). Only discovery may + // overlap. Self-host runs a single libSQL connection issuing raw + // BEGIN/COMMIT, where a second transaction opened while one is live fails + // outright with "cannot start a transaction within a transaction" — the + // failure #1563 fixed for concurrent refreshes of the SAME connection via + // the single-flight map below. Rebuilding several DIFFERENT connections + // together (the stale-catalog fan-out) reopens the same hazard from the + // other side, so the write phase takes a single permit: the fan-out's + // discoveries still run together and their commits form a queue. + // + // Never take this permit while a transaction is already open on this fiber + // — every caller of `persistCatalog` must be outside one, as all of the + // `produceConnectionTools` call sites are. + const catalogPersistLock = Semaphore.makeUnsafe(1); + const persistCatalog = (effect: Effect.Effect) => + catalogPersistLock.withPermits(1)(transaction(effect)); + const produceConnectionToolsUnshared = ( integrationRow: IntegrationRow, ref: ConnectionRef, @@ -2941,7 +2993,7 @@ export const createExecutor = [] = []; for (const connection of connections) { const integrationRow = integrationBySlug.get(connection.integration); if (!integrationRow) continue; @@ -4066,24 +4125,38 @@ export const createExecutor = Effect.succeed([] as readonly Tool[])), - Effect.withSpan("executor.tools.sync_stale", { - attributes: { - "executor.integration": connection.integration, - "executor.connection": connection.name, + rebuilds.push( + produceConnectionTools( + integrationRow, + { + owner: connection.owner as Owner, + integration: IntegrationSlug.make(connection.integration), + name: ConnectionName.make(connection.name), }, - }), + "background", + ).pipe( + // Best-effort, but never silent: the read still succeeds on the + // stale-but-working catalog and the peer rebuilds still finish, + // while the operator gets the connection that failed and why. + // Without this a connection whose upstream is permanently broken + // re-fails on every read and leaves no trace anywhere. + Effect.catch((error) => + Effect.logWarning("executor stale tool sync failed", { + integration: connection.integration, + connection: connection.name, + error: describeSyncFailure(error), + }).pipe(Effect.as([] as readonly Tool[])), + ), + Effect.withSpan("executor.tools.sync_stale", { + attributes: { + "executor.integration": connection.integration, + "executor.connection": connection.name, + }, + }), + ), ); } + yield* Effect.all(rebuilds, { concurrency: STALE_TOOLS_SYNC_CONCURRENCY }); }); const toolsList = (filter?: ToolListFilter): Effect.Effect => diff --git a/packages/core/sdk/src/index.ts b/packages/core/sdk/src/index.ts index aa241f371f..107ff5fce7 100644 --- a/packages/core/sdk/src/index.ts +++ b/packages/core/sdk/src/index.ts @@ -422,6 +422,7 @@ export { type ExecutorDbFactory, type ExecutorDbInput, type ParsedToolAddress, + STALE_TOOLS_SYNC_CONCURRENCY, createExecutor, collectTables, parseToolAddress, diff --git a/packages/plugins/mcp/src/sdk/catalog-sync.test.ts b/packages/plugins/mcp/src/sdk/catalog-sync.test.ts index 88c37ab5b8..03dd2a6e45 100644 --- a/packages/plugins/mcp/src/sdk/catalog-sync.test.ts +++ b/packages/plugins/mcp/src/sdk/catalog-sync.test.ts @@ -14,13 +14,14 @@ // --------------------------------------------------------------------------- import { describe, expect, it } from "@effect/vitest"; -import { Effect, Option, Schema } from "effect"; +import { Deferred, Effect, Fiber, Option, Ref, Schema } from "effect"; import { HttpServerResponse } from "effect/unstable/http"; import { AuthTemplateSlug, ConnectionName, IntegrationSlug, + STALE_TOOLS_SYNC_CONCURRENCY, ToolAddress, createExecutor, } from "@executor-js/sdk"; @@ -267,3 +268,136 @@ describe("MCP tools/list pagination", () => { }), ); }); + +// --------------------------------------------------------------------------- +// Stale-catalog refresh concurrency. +// +// A tools read rebuilds every stale connection it finds. Each rebuild is an +// independent upstream listing, so a host with several stale remote catalogs +// must not pay the sum of every server's latency on the read that trips the +// TTL. Nor may one read open an unbounded number of upstream listings. +// +// 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. +// --------------------------------------------------------------------------- + +const STALE_CONNECTIONS = STALE_TOOLS_SYNC_CONCURRENCY + 1; + +const serveLatchedListServer = () => + Effect.gen(function* () { + const armed = yield* Ref.make(false); + const listings = yield* Ref.make(0); + // Signalled when the bound is saturated; released by the test, not by the + // fixture, so the test can first prove nothing beyond the bound arrives. + const atLimit = yield* Deferred.make(); + 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-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 the test releases them. A serial + // 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) { + yield* Deferred.succeed(atLimit, undefined); + } + yield* Deferred.await(release); + } + return jsonRpcResult(rpc, { tools: [pageTool("alpha")] }); + }), + ); + + return { + // Distinct endpoint paths so each connection dials its own MCP session + // instead of sharing one pooled client. + endpoint: (index: number) => server.url(`/mcp/${index}`), + arm: Ref.set(armed, true), + awaitLimit: Deferred.await(atLimit), + release: Deferred.succeed(release, undefined), + listings: Ref.get(listings), + } as const; + }); + +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, + }); + + 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); + + // 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); + }), + ); +});