From 5c37b310602ccbd44c3ab9e4d99897eba95645fc Mon Sep 17 00:00:00 2001 From: Aditya kumar singh <143548997+Adityakk9031@users.noreply.github.com> Date: Mon, 10 Aug 2026 17:27:38 +0530 Subject: [PATCH 1/4] fix(selfhost): refresh stale connection tools concurrently and expose EXECUTOR_TOOLS_SYNC_TTL_MS --- apps/host-selfhost/src/config.ts | 11 ++++++ apps/host-selfhost/src/execution.ts | 1 + .../host-selfhost/src/executor-config.test.ts | 25 +++++++++++++ .../core/api/src/server/scoped-executor.ts | 6 ++++ packages/core/sdk/src/executor.ts | 36 +++++++++++-------- 5 files changed, 64 insertions(+), 15 deletions(-) diff --git a/apps/host-selfhost/src/config.ts b/apps/host-selfhost/src/config.ts index e0cd282d52..866981516d 100644 --- a/apps/host-selfhost/src/config.ts +++ b/apps/host-selfhost/src/config.ts @@ -43,6 +43,8 @@ export interface SelfHostConfig { readonly organizationName: string; /** URL slug for org-prefixed console paths (`//policies`). */ readonly orgSlug: string; + /** Freshness TTL (in ms) for remote tool catalogs, or `null` to disable. */ + readonly toolsSyncTtlMs?: number | null; } export const resolveDataDir = (): string => @@ -148,6 +150,7 @@ export const loadConfig = (): SelfHostConfig => { bootstrapAdminName: process.env.EXECUTOR_BOOTSTRAP_ADMIN_NAME ?? "Admin", organizationName: process.env.EXECUTOR_ORG_NAME ?? "Default", orgSlug: resolveOrgSlug(), + toolsSyncTtlMs: resolveToolsSyncTtlMs(), }; }; @@ -165,3 +168,11 @@ const resolveOrgSlug = (): string => { } return slug; }; + +const resolveToolsSyncTtlMs = (): number | null | undefined => { + const raw = process.env.EXECUTOR_TOOLS_SYNC_TTL_MS?.trim(); + if (!raw) return undefined; + if (raw === "null" || raw === "false" || raw === "0") return null; + const parsed = Number.parseInt(raw, 10); + return Number.isNaN(parsed) ? undefined : parsed; +}; diff --git a/apps/host-selfhost/src/execution.ts b/apps/host-selfhost/src/execution.ts index 270ffc4f8e..d196dd215d 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..7598aade82 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,20 @@ test("stdio MCP is enabled when the opt-in is exactly true", () => { process.env[ENV_NAME] = "true"; expect(allowStdio()).toBe(true); }); + +test("toolsSyncTtlMs parses integer, null/false/0 disable values, and undefined fallback", () => { + delete process.env[TTL_ENV_NAME]; + expect(loadConfig().toolsSyncTtlMs).toBeUndefined(); + + process.env[TTL_ENV_NAME] = "60000"; + expect(loadConfig().toolsSyncTtlMs).toBe(60000); + + process.env[TTL_ENV_NAME] = "null"; + expect(loadConfig().toolsSyncTtlMs).toBeNull(); + + process.env[TTL_ENV_NAME] = "false"; + expect(loadConfig().toolsSyncTtlMs).toBeNull(); + + process.env[TTL_ENV_NAME] = "0"; + expect(loadConfig().toolsSyncTtlMs).toBeNull(); +}); diff --git a/packages/core/api/src/server/scoped-executor.ts b/packages/core/api/src/server/scoped-executor.ts index ea0e33ce61..25aae932a8 100644 --- a/packages/core/api/src/server/scoped-executor.ts +++ b/packages/core/api/src/server/scoped-executor.ts @@ -97,6 +97,11 @@ export interface HostConfigShape { * Hosts that record product analytics supply it; omitted -> no observation. */ readonly onIntegrationChange?: ExecutorConfig["onIntegrationChange"]; + /** + * Freshness TTL (in ms) for remote tool catalogs before an explicit re-sync is + * attempted. Omit for default (15 mins), or set `null` to disable time-based re-sync. + */ + readonly toolsSyncTtlMs?: number | null; } export class HostConfig extends Context.Service()( @@ -284,6 +289,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/executor.ts b/packages/core/sdk/src/executor.ts index 56cb2e2997..66f005d0ff 100644 --- a/packages/core/sdk/src/executor.ts +++ b/packages/core/sdk/src/executor.ts @@ -3631,6 +3631,7 @@ export const createExecutor = Effect.succeed([] as readonly Tool[])), - Effect.withSpan("executor.tools.sync_stale", { - attributes: { - "executor.integration": connection.integration, - "executor.connection": connection.name, + tasks.push( + produceConnectionTools( + integrationRow, + { + owner: connection.owner as Owner, + integration: IntegrationSlug.make(connection.integration), + name: ConnectionName.make(connection.name), }, - }), + "background", + ).pipe( + Effect.catch(() => Effect.succeed([] as readonly Tool[])), + Effect.withSpan("executor.tools.sync_stale", { + attributes: { + "executor.integration": connection.integration, + "executor.connection": connection.name, + }, + }), + ), ); } + if (tasks.length > 0) { + yield* Effect.all(tasks, { concurrency: 10 }); + } }); const toolsList = (filter?: ToolListFilter): Effect.Effect => From 8196a1fd9c0bee2ba5380f08155f73c5067e2f97 Mon Sep 17 00:00:00 2001 From: Rhys Sullivan <39114868+RhysSullivan@users.noreply.github.com> Date: Thu, 27 Aug 2026 21:35:08 -0700 Subject: [PATCH 2/4] Refuse a malformed tools-sync TTL and cover the concurrent refresh Fail at boot on an unparseable or negative EXECUTOR_TOOLS_SYNC_TTL_MS instead of falling back to the default. 0 keeps meaning disabled, now spelled out in the env var's docs and mapped explicitly onto the SDK's null sentinel, since a TTL of 0 means the opposite to the SDK. Type the rebuild list, name the concurrency bound, and add the regression test the concurrency change was missing: a latched MCP fixture that only answers once the whole stale set is in flight, so a serial refresh cannot complete. --- .changeset/concurrent-stale-tools-sync.md | 9 ++ apps/host-selfhost/src/config.ts | 39 ++++++- .../host-selfhost/src/executor-config.test.ts | 31 ++++- packages/core/sdk/src/executor.ts | 17 ++- .../plugins/mcp/src/sdk/catalog-sync.test.ts | 107 +++++++++++++++++- 5 files changed, 186 insertions(+), 17 deletions(-) create mode 100644 .changeset/concurrent-stale-tools-sync.md diff --git a/.changeset/concurrent-stale-tools-sync.md b/.changeset/concurrent-stale-tools-sync.md new file mode 100644 index 0000000000..7b483bfa41 --- /dev/null +++ b/.changeset/concurrent-stale-tools-sync.md @@ -0,0 +1,9 @@ +--- +"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. They now run concurrently, bounded so a large stale set cannot open an unbounded number of listings from one read. + +Self-host also exposes the freshness window as `EXECUTOR_TOOLS_SYNC_TTL_MS`. Leave it unset for the 15-minute default, or set `0` (equivalently `off`, `null` or `false`) to disable time-based re-sync and leave stale-marking and config revision as the only refresh triggers. A malformed or negative 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 38d18d4216..b76ef816e3 100644 --- a/apps/host-selfhost/src/config.ts +++ b/apps/host-selfhost/src/config.ts @@ -51,8 +51,12 @@ export interface SelfHostConfig { * minutes (the same pattern as MCP_PAUSED_SESSION_IDLE_TIMEOUT_MS on cloud). */ readonly sandboxTimeoutMs: number | undefined; - /** Freshness TTL (in ms) for remote tool catalogs, or `null` to disable. */ - readonly toolsSyncTtlMs?: number | null; + /** + * 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 => @@ -194,10 +198,35 @@ 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. +// +// `0` disables time-based re-sync, and is mapped to the SDK's `null` sentinel +// rather than forwarded: to the SDK a TTL of 0 means the opposite — every +// catalog is expired on every read. "off", "null" and "false" spell the same +// disable, since operators reach for all three. +// +// 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 resolveToolsSyncTtlMs = (): number | null | undefined => { const raw = process.env.EXECUTOR_TOOLS_SYNC_TTL_MS?.trim(); if (!raw) return undefined; - if (raw === "null" || raw === "false" || raw === "0") return null; - const parsed = Number.parseInt(raw, 10); - return Number.isNaN(parsed) ? undefined : parsed; + if (raw === "off" || raw === "null" || raw === "false") return null; + const parsed = Number(raw); + if (!Number.isInteger(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 a whole number of milliseconds ("0", "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 "0" to disable time-based re-sync)`, + ); + } + return parsed === 0 ? null : parsed; }; diff --git a/apps/host-selfhost/src/executor-config.test.ts b/apps/host-selfhost/src/executor-config.test.ts index 7598aade82..7714b46887 100644 --- a/apps/host-selfhost/src/executor-config.test.ts +++ b/apps/host-selfhost/src/executor-config.test.ts @@ -66,19 +66,38 @@ test("stdio MCP is enabled when the opt-in is exactly true", () => { expect(allowStdio()).toBe(true); }); -test("toolsSyncTtlMs parses integer, null/false/0 disable values, and undefined fallback", () => { +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); +}); - process.env[TTL_ENV_NAME] = "null"; +// 0 is the operator-facing way to turn the TTL off. It is deliberately NOT +// forwarded as 0, which the SDK reads as "expired on every read" — the exact +// opposite — so the resolver maps it onto the SDK's `null` disable sentinel. +test.each(["0", "off", "null", "false"])("the tools-sync TTL is disabled by %s", (raw) => { + process.env[TTL_ENV_NAME] = raw; expect(loadConfig().toolsSyncTtlMs).toBeNull(); +}); - process.env[TTL_ENV_NAME] = "false"; - 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. +test.each(["abc", "60_000", "1.5", "1e3ms", "NaN", "Infinity"])( + "a malformed tools-sync TTL (%s) refuses to boot", + (raw) => { + process.env[TTL_ENV_NAME] = raw; + expect(() => loadConfig()).toThrow(/EXECUTOR_TOOLS_SYNC_TTL_MS/); + }, +); - process.env[TTL_ENV_NAME] = "0"; - expect(loadConfig().toolsSyncTtlMs).toBeNull(); +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/sdk/src/executor.ts b/packages/core/sdk/src/executor.ts index 18e2673441..80c96e3d6b 100644 --- a/packages/core/sdk/src/executor.ts +++ b/packages/core/sdk/src/executor.ts @@ -715,6 +715,11 @@ export interface ExecutorConfig[] = []; for (const connection of connections) { const integrationRow = integrationBySlug.get(connection.integration); if (!integrationRow) continue; @@ -4067,7 +4076,7 @@ export const createExecutor = 0) { - yield* Effect.all(tasks, { concurrency: 10 }); - } + yield* Effect.all(rebuilds, { concurrency: STALE_TOOLS_SYNC_CONCURRENCY }); }); const toolsList = (filter?: ToolListFilter): Effect.Effect => diff --git a/packages/plugins/mcp/src/sdk/catalog-sync.test.ts b/packages/plugins/mcp/src/sdk/catalog-sync.test.ts index 88c37ab5b8..1aac2c79aa 100644 --- a/packages/plugins/mcp/src/sdk/catalog-sync.test.ts +++ b/packages/plugins/mcp/src/sdk/catalog-sync.test.ts @@ -14,7 +14,7 @@ // --------------------------------------------------------------------------- import { describe, expect, it } from "@effect/vitest"; -import { Effect, Option, Schema } from "effect"; +import { Deferred, Effect, Option, Ref, Schema } from "effect"; import { HttpServerResponse } from "effect/unstable/http"; import { @@ -267,3 +267,108 @@ 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. This is the regression guard for that: the fixture below refuses to +// answer any listing until all of them are in flight together, so a serial +// refresh cannot finish at all. +// --------------------------------------------------------------------------- + +const STALE_CONNECTIONS = 4; + +const serveLatchedListServer = () => + Effect.gen(function* () { + const armed = yield* Ref.make(false); + const listings = yield* Ref.make(0); + const allInFlight = 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 whole stale set has arrived. + // Serial refresh parks on the first one forever. + if (yield* Ref.get(armed)) { + const arrived = yield* Ref.updateAndGet(listings, (n) => n + 1); + if (arrived >= STALE_CONNECTIONS) yield* Deferred.succeed(allInFlight, undefined); + yield* Deferred.await(allInFlight); + } + 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), + listings: Ref.get(listings), + } as const; + }); + +describe("MCP stale-catalog refresh", () => { + it.effect("rebuilds every stale connection concurrently, not one after another", () => + 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; + + // Well inside the harness timeout, so a serial refresh fails on the + // assertion below rather than as an opaque test-runner timeout. + const refreshed = yield* executor.tools.list().pipe(Effect.timeoutOption("10 seconds")); + + // A serial refresh never releases the latch, so the read times out here. + expect(Option.isSome(refreshed)).toBe(true); + expect(yield* fixture.listings).toBe(STALE_CONNECTIONS); + }), + ); +}); From 25d06ad34a66c7b636d3ef427c04273042f10d5a Mon Sep 17 00:00:00 2001 From: Rhys Sullivan <39114868+RhysSullivan@users.noreply.github.com> Date: Thu, 27 Aug 2026 21:38:05 -0700 Subject: [PATCH 3/4] Forward a zero TTL to the SDK instead of remapping it to disabled --- .changeset/concurrent-stale-tools-sync.md | 2 +- apps/host-selfhost/src/config.ts | 14 +++++++------- apps/host-selfhost/src/executor-config.test.ts | 12 ++++++++---- 3 files changed, 16 insertions(+), 12 deletions(-) diff --git a/.changeset/concurrent-stale-tools-sync.md b/.changeset/concurrent-stale-tools-sync.md index 7b483bfa41..02f8c7a623 100644 --- a/.changeset/concurrent-stale-tools-sync.md +++ b/.changeset/concurrent-stale-tools-sync.md @@ -6,4 +6,4 @@ 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. They now run concurrently, bounded so a large stale set cannot open an unbounded number of listings from one read. -Self-host also exposes the freshness window as `EXECUTOR_TOOLS_SYNC_TTL_MS`. Leave it unset for the 15-minute default, or set `0` (equivalently `off`, `null` or `false`) to disable time-based re-sync and leave stale-marking and config revision as the only refresh triggers. A malformed or negative value is refused at boot rather than silently falling back to the default. +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`) 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 or negative 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 b76ef816e3..2934875edd 100644 --- a/apps/host-selfhost/src/config.ts +++ b/apps/host-selfhost/src/config.ts @@ -203,10 +203,10 @@ const resolveOrgSlug = (): string => { // fresh before the next tools read re-lists it. Unset takes the SDK default of // 15 minutes. // -// `0` disables time-based re-sync, and is mapped to the SDK's `null` sentinel -// rather than forwarded: to the SDK a TTL of 0 means the opposite — every -// catalog is expired on every read. "off", "null" and "false" spell the same -// disable, since operators reach for all three. +// 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. // // 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 @@ -219,14 +219,14 @@ const resolveToolsSyncTtlMs = (): number | null | undefined => { if (!Number.isInteger(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 a whole number of milliseconds ("0", "off", "null" or "false" disable time-based re-sync)`, + `EXECUTOR_TOOLS_SYNC_TTL_MS ${JSON.stringify(raw)} is not a 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 "0" to disable time-based re-sync)`, + `EXECUTOR_TOOLS_SYNC_TTL_MS ${JSON.stringify(raw)} must not be negative (use "off" to disable time-based re-sync)`, ); } - return parsed === 0 ? null : parsed; + return parsed; }; diff --git a/apps/host-selfhost/src/executor-config.test.ts b/apps/host-selfhost/src/executor-config.test.ts index 7714b46887..09bf816eba 100644 --- a/apps/host-selfhost/src/executor-config.test.ts +++ b/apps/host-selfhost/src/executor-config.test.ts @@ -79,10 +79,14 @@ test("a positive tools-sync TTL is forwarded verbatim", () => { expect(loadConfig().toolsSyncTtlMs).toBe(60000); }); -// 0 is the operator-facing way to turn the TTL off. It is deliberately NOT -// forwarded as 0, which the SDK reads as "expired on every read" — the exact -// opposite — so the resolver maps it onto the SDK's `null` disable sentinel. -test.each(["0", "off", "null", "false"])("the tools-sync TTL is disabled by %s", (raw) => { +// 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); +}); + +test.each(["off", "null", "false"])("the tools-sync TTL is disabled by %s", (raw) => { process.env[TTL_ENV_NAME] = raw; expect(loadConfig().toolsSyncTtlMs).toBeNull(); }); From f051090f4c94642d15c303bff955d0ca771b630b Mon Sep 17 00:00:00 2001 From: Rhys Sullivan <39114868+RhysSullivan@users.noreply.github.com> Date: Thu, 27 Aug 2026 23:38:13 -0700 Subject: [PATCH 4/4] Serialize catalog writes in the stale tools fan-out and log failed rebuilds The stale-catalog refresh rebuilds several connections at once. Each rebuild ends in a catalog-replacement transaction, and a self-host database is one connection issuing raw BEGIN/COMMIT, so overlapping rebuilds could reopen the transaction collision fixed for same-connection refreshes. Split the phases: upstream discovery stays concurrent at the existing bound, while every catalog write takes a single permit and commits in turn. A rebuild that fails now logs a warning with the connection and the reason, including the cause a plain structural render would drop. The read still succeeds on the stale-but-working catalog and the other rebuilds still finish. Refuse a tools-sync TTL that is not a safe integer, since a larger value silently rounds, and accept the disable tokens in any case. Tests cover the concurrency bound (an extra connection waits while the bound is saturated), non-overlapping persistence observed through real transactions, and a failed rebuild that warns without failing the read. --- .changeset/concurrent-stale-tools-sync.md | 6 +- apps/host-selfhost/src/config.ts | 14 +- .../host-selfhost/src/executor-config.test.ts | 17 +- packages/core/sdk/src/connections.test.ts | 239 +++++++++++++++++- packages/core/sdk/src/executor.ts | 78 +++++- packages/core/sdk/src/index.ts | 1 + .../plugins/mcp/src/sdk/catalog-sync.test.ts | 59 +++-- 7 files changed, 378 insertions(+), 36 deletions(-) diff --git a/.changeset/concurrent-stale-tools-sync.md b/.changeset/concurrent-stale-tools-sync.md index 02f8c7a623..8f4bc858ca 100644 --- a/.changeset/concurrent-stale-tools-sync.md +++ b/.changeset/concurrent-stale-tools-sync.md @@ -4,6 +4,8 @@ **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. They now run concurrently, bounded so a large stale set cannot open an unbounded number of listings from one read. +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. -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`) 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 or negative value is refused at boot rather than silently falling back to the default. +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 2934875edd..acb77204f3 100644 --- a/apps/host-selfhost/src/config.ts +++ b/apps/host-selfhost/src/config.ts @@ -206,20 +206,26 @@ const resolveOrgSlug = (): string => { // 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. +// 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 (raw === "off" || raw === "null" || raw === "false") return null; + if (TOOLS_SYNC_TTL_DISABLE_TOKENS.has(raw.toLowerCase())) return null; const parsed = Number(raw); - if (!Number.isInteger(parsed)) { + // `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 a whole number of milliseconds ("off", "null" or "false" disable time-based re-sync)`, + `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) { diff --git a/apps/host-selfhost/src/executor-config.test.ts b/apps/host-selfhost/src/executor-config.test.ts index 09bf816eba..313d097b85 100644 --- a/apps/host-selfhost/src/executor-config.test.ts +++ b/apps/host-selfhost/src/executor-config.test.ts @@ -86,14 +86,21 @@ test("a zero tools-sync TTL forwards as the SDK's always-stale 0", () => { expect(loadConfig().toolsSyncTtlMs).toBe(0); }); -test.each(["off", "null", "false"])("the tools-sync TTL is disabled by %s", (raw) => { - process.env[TTL_ENV_NAME] = raw; - expect(loadConfig().toolsSyncTtlMs).toBeNull(); -}); +// 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. -test.each(["abc", "60_000", "1.5", "1e3ms", "NaN", "Infinity"])( +// "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; 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 80c96e3d6b..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,10 +725,11 @@ 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>({ @@ -2879,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, @@ -2946,7 +2993,7 @@ export const createExecutor = [] = []; for (const connection of connections) { const integrationRow = integrationBySlug.get(connection.integration); @@ -4086,7 +4135,18 @@ export const createExecutor = Effect.succeed([] as readonly Tool[])), + // 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, 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 1aac2c79aa..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 { Deferred, Effect, Option, Ref, 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"; @@ -274,18 +275,26 @@ describe("MCP tools/list pagination", () => { // 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. This is the regression guard for that: the fixture below refuses to -// answer any listing until all of them are in flight together, so a serial -// refresh cannot finish at all. +// 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 = 4; +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); - const allInFlight = yield* Deferred.make(); + // 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* () { @@ -310,12 +319,14 @@ const serveLatchedListServer = () => if (rpc.method !== "tools/list") { return HttpServerResponse.text("Unexpected JSON-RPC method", { status: 400 }); } - // Once armed, park every listing until the whole stale set has arrived. - // Serial refresh parks on the first one forever. + // 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_CONNECTIONS) yield* Deferred.succeed(allInFlight, undefined); - yield* Deferred.await(allInFlight); + if (arrived >= STALE_TOOLS_SYNC_CONCURRENCY) { + yield* Deferred.succeed(atLimit, undefined); + } + yield* Deferred.await(release); } return jsonRpcResult(rpc, { tools: [pageTool("alpha")] }); }), @@ -326,12 +337,17 @@ const serveLatchedListServer = () => // 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.effect("rebuilds every stale connection concurrently, not one after another", () => + // `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({ @@ -362,11 +378,24 @@ describe("MCP stale-catalog refresh", () => { yield* executor.tools.list(); yield* fixture.arm; - // Well inside the harness timeout, so a serial refresh fails on the - // assertion below rather than as an opaque test-runner timeout. - const refreshed = yield* executor.tools.list().pipe(Effect.timeoutOption("10 seconds")); + 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); - // A serial refresh never releases the latch, so the read times out here. + // 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); }),