From 4966fe7d1fc07158762b2740b07917d65d6764ba Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Tue, 20 Oct 2026 21:07:26 -0700 Subject: [PATCH 1/2] fix(mothership): bound best-effort mcp discovery and stop retrying a down server every query Broad chat catalog discovery started best-effort discovery for every usable workspace server at once. The 3s wait only ends the caller's wait, so each attempt kept its sockets open for up to ~2 min, and a workspace with many servers could burst past the connection pool and evict other users' warm connections. Best-effort failures write no shared cooldown, so a down server was reattempted (with its DB and Redis reads) on every query. - Run best-effort discovery through mapSettledWithConcurrency with 8 at once; each server's 3s wait counts from its own start, and servers still queued 3s into the query are skipped unstarted, bounding query latency at ~6s. - Remember, per process and per requester (workspace:server:user, 60s, capped at 1000 entries), a best-effort server that failed or outlived its wait, and skip it until the memo expires or the late discovery succeeds. Nothing is written to shared state. - Named servers are unchanged: listed at once, waited on to completion. --- apps/sim/lib/mothership/mcp-tools.test.ts | 159 +++++++++++++++++++++- apps/sim/lib/mothership/mcp-tools.ts | 90 ++++++++++-- 2 files changed, 234 insertions(+), 15 deletions(-) diff --git a/apps/sim/lib/mothership/mcp-tools.test.ts b/apps/sim/lib/mothership/mcp-tools.test.ts index 1da2df8302d..7fbdec49d34 100644 --- a/apps/sim/lib/mothership/mcp-tools.test.ts +++ b/apps/sim/lib/mothership/mcp-tools.test.ts @@ -1,5 +1,6 @@ +import { flushMicrotasks } from '@sim/testing' import { mcpUseCasesMock, mcpUseCasesMockFns } from '@sim/testing/mocks/mcp-use-cases.mock' -import { describe, expect, it, vi } from 'vitest' +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' const { discoverServerTools } = vi.hoisted(() => ({ discoverServerTools: vi.fn(), @@ -10,6 +11,7 @@ vi.mock('@/lib/internal/mcp/discover-tools', () => ({ })) vi.mock('@/lib/mcp/application/use-cases', () => mcpUseCasesMock) +import { DeadlineExceededError } from '@/lib/core/utils/deadline' import { buildMcpToolSchemas } from '@/lib/mothership/mcp-tools' mcpUseCasesMockFns.mockDiscoverMcpServerToolsUseCase.mockImplementation(async (args: unknown) => ({ @@ -81,3 +83,158 @@ describe('mothership MCP tool schemas', () => { expect(discovery.failures).toEqual([]) }) }) + +/** + * Broad chat discovery reaches every usable workspace server. Its failure modes are fan-out + * (every server's connect started at once), latency that grows with the server count, a slow + * server cut short by a wait that began before it started, and a down server reattempted on + * every query. None of it may touch the servers the request names. + */ +describe('mothership MCP best-effort discovery bounds', () => { + const hanging = () => new Promise(() => undefined) + const toolsOf = (serverId: string) => [{ serverId, name: 'search', inputSchema: {} }] + const serverIdOf = (args: unknown) => (args as { input: { serverId: string } }).input.serverId + const startedServers = () => discoverServerTools.mock.calls.map(([args]) => serverIdOf(args)) + const ids = (prefix: string, count: number) => + Array.from({ length: count }, (_, index) => `${prefix}-${index + 1}`) + + beforeEach(() => { + vi.useFakeTimers() + discoverServerTools.mockReset() + }) + afterEach(() => { + vi.useRealTimers() + }) + + it('starts at most eight best-effort discoveries at once and skips the rest once the start window closes', async () => { + discoverServerTools.mockImplementation(hanging) + const servers = ids('mcp-hang', 20) + + const pending = buildMcpToolSchemas('user-cap', 'ws-cap', { complete: [], bestEffort: servers }) + await flushMicrotasks(10) + expect(startedServers()).toEqual(servers.slice(0, 8)) + + await vi.advanceTimersByTimeAsync(3_000) + const discovery = await pending + expect(startedServers()).toEqual(servers.slice(0, 8)) + expect(discovery.failures.map(({ serverId }) => serverId)).toEqual(servers) + expect(discovery.failures.every(({ error }) => error instanceof DeadlineExceededError)).toBe( + true + ) + }) + + it('keeps discovering fast servers while slow ones hold their slots', async () => { + const fast = ids('mcp-fast', 10) + discoverServerTools.mockImplementation(async (args: unknown) => { + const serverId = serverIdOf(args) + return serverId === 'mcp-slow' ? hanging() : toolsOf(serverId) + }) + + const pending = buildMcpToolSchemas('user-fast', 'ws-fast', { + complete: [], + bestEffort: ['mcp-slow', ...fast], + }) + await flushMicrotasks(50) + expect(startedServers()).toEqual(['mcp-slow', ...fast]) + + await vi.advanceTimersByTimeAsync(3_000) + const discovery = await pending + expect(discovery.tools.map((tool) => tool.service)).toEqual( + fast.map((serverId) => `mcp:${serverId}`) + ) + expect(discovery.failures.map(({ serverId }) => serverId)).toEqual(['mcp-slow']) + }) + + it("measures each server's wait from its own start, not the query's", async () => { + const blockers = ids('mcp-block', 8) + discoverServerTools.mockImplementation((args: unknown) => { + const serverId = serverIdOf(args) + if (serverId === 'mcp-block-1') + return new Promise((resolve) => setTimeout(() => resolve(toolsOf(serverId)), 1_000)) + if (serverId === 'mcp-late') { + const resolveAt = 3_500 - (Date.now() - startedAt) + return new Promise((resolve) => setTimeout(() => resolve(toolsOf(serverId)), resolveAt)) + } + return hanging() + }) + const startedAt = Date.now() + + const pending = buildMcpToolSchemas('user-own', 'ws-own', { + complete: [], + bestEffort: [...blockers, 'mcp-late'], + }) + await vi.advanceTimersByTimeAsync(4_000) + const discovery = await pending + + expect(discovery.tools.map((tool) => tool.service)).toEqual(['mcp:mcp-block-1', 'mcp:mcp-late']) + expect(discovery.failures.map(({ serverId }) => serverId)).toEqual(blockers.slice(1)) + }) + + it('lists every named server at once and waits for each however long it takes', async () => { + const named = ids('mcp-named', 12) + discoverServerTools.mockImplementation((args: unknown) => { + const serverId = serverIdOf(args) + return serverId.startsWith('mcp-named') + ? new Promise((resolve) => setTimeout(() => resolve(toolsOf(serverId)), 30_000)) + : hanging() + }) + + const pending = buildMcpToolSchemas('user-named', 'ws-named', { + complete: named, + bestEffort: ids('mcp-optional', 10), + }) + await flushMicrotasks(10) + expect(startedServers()).toEqual([...named, ...ids('mcp-optional', 8)]) + + await vi.advanceTimersByTimeAsync(30_000) + const discovery = await pending + expect(discovery.tools.map((tool) => tool.service)).toEqual( + named.map((serverId) => `mcp:${serverId}`) + ) + expect(discovery.failures.map(({ serverId }) => serverId)).toEqual(ids('mcp-optional', 10)) + }) + + it('does not reattempt a failed best-effort server for the same requester until the memo expires', async () => { + discoverServerTools.mockRejectedValue(new Error('connect ECONNREFUSED')) + const query = (userId: string) => + buildMcpToolSchemas(userId, 'ws-memo', { complete: [], bestEffort: ['mcp-down'] }) + + expect((await query('user-memo-a')).failures.map(({ error }) => error.message)).toEqual([ + 'connect ECONNREFUSED', + ]) + expect((await query('user-memo-a')).failures.map(({ error }) => error.message)).toEqual([ + 'connect ECONNREFUSED', + ]) + expect(discoverServerTools).toHaveBeenCalledTimes(1) + + await query('user-memo-b') + expect(discoverServerTools).toHaveBeenCalledTimes(2) + + vi.advanceTimersByTime(60_000) + await query('user-memo-a') + expect(discoverServerTools).toHaveBeenCalledTimes(3) + }) + + it('skips a server whose discovery outlived the wait until that discovery settles, then serves it', async () => { + discoverServerTools.mockImplementation( + (args: unknown) => + new Promise((resolve) => setTimeout(() => resolve(toolsOf(serverIdOf(args))), 5_000)) + ) + const query = () => + buildMcpToolSchemas('user-slow', 'ws-slow', { complete: [], bestEffort: ['mcp-slow'] }) + + const first = query() + await vi.advanceTimersByTimeAsync(3_000) + expect((await first).failures.map(({ serverId }) => serverId)).toEqual(['mcp-slow']) + + const second = query() + await flushMicrotasks(10) + expect(discoverServerTools).toHaveBeenCalledTimes(1) + expect((await second).failures.map(({ serverId }) => serverId)).toEqual(['mcp-slow']) + + await vi.advanceTimersByTimeAsync(2_000) + discoverServerTools.mockResolvedValueOnce(toolsOf('mcp-slow')) + expect((await query()).tools.map((tool) => tool.service)).toEqual(['mcp:mcp-slow']) + expect(discoverServerTools).toHaveBeenCalledTimes(2) + }) +}) diff --git a/apps/sim/lib/mothership/mcp-tools.ts b/apps/sim/lib/mothership/mcp-tools.ts index d0099273b4e..5e9aa9047b0 100644 --- a/apps/sim/lib/mothership/mcp-tools.ts +++ b/apps/sim/lib/mothership/mcp-tools.ts @@ -1,5 +1,6 @@ import { toError } from '@sim/utils/errors' -import { withinDeadline } from '@/lib/core/utils/deadline' +import { mapSettledWithConcurrency } from '@/lib/core/utils/concurrency' +import { DeadlineExceededError, withinDeadline } from '@/lib/core/utils/deadline' import { discoverMcpServerToolsAsExecutor } from '@/lib/internal/mcp/discover-tools' import type { InternalToolOperationContext } from '@/lib/internal/tool-operations/types' import { MCP_SERVER_DELEGATION_AUDIENCE } from '@/lib/mcp/application/authorization' @@ -10,11 +11,51 @@ import { createCopilotChatPrincipal } from '@/lib/mothership/auth/application-de import type { ToolSchema } from '@/lib/mothership/chat/payload' /** - * How long one best-effort server may hold up a catalog query. A warm server is - * a cache read; a cold or slow one is skipped for this query while its - * discovery keeps running and fills the cache for the next one. + * How long a catalog query waits on one best-effort server, counted from that + * server's own start. A warm server is a cache read; a cold or slow one is + * skipped for this query while its discovery keeps running and fills the cache + * for the next one. */ const BEST_EFFORT_DISCOVERY_WAIT_MS = 3_000 +/** + * Best-effort discoveries a query runs at once. The wait does not cancel a + * discovery, so this also caps the connections one query can leave running. + */ +const BEST_EFFORT_DISCOVERY_CONCURRENCY = 8 +/** + * How long after a query begins it may still start a best-effort discovery. + * Servers still queued then are skipped for this query, so with the wait it + * bounds the query's latency at about six seconds however many servers the + * workspace has. + */ +const BEST_EFFORT_DISCOVERY_START_WINDOW_MS = 3_000 +/** How long a requester skips a best-effort server whose discovery failed or outlived the wait. */ +const BEST_EFFORT_RETRY_AFTER_MS = 60_000 +const MAX_RECENT_BEST_EFFORT_FAILURES = 1_000 + +/** + * Best-effort servers that recently failed for a requester, keyed by + * `workspaceId:serverId:userId`, oldest first. It lives in this process because + * best-effort discovery must leave the server's shared status and cooldown to + * the discoveries someone asked for. + */ +const recentBestEffortFailures = new Map() + +function recentBestEffortFailure(key: string): Error | undefined { + const failure = recentBestEffortFailures.get(key) + if (!failure) return undefined + if (failure.retryAfter > Date.now()) return failure.error + recentBestEffortFailures.delete(key) + return undefined +} + +function rememberBestEffortFailure(key: string, error: Error): void { + recentBestEffortFailures.delete(key) + recentBestEffortFailures.set(key, { error, retryAfter: Date.now() + BEST_EFFORT_RETRY_AFTER_MS }) + if (recentBestEffortFailures.size <= MAX_RECENT_BEST_EFFORT_FAILURES) return + const oldest = recentBestEffortFailures.keys().next() + if (!oldest.done) recentBestEffortFailures.delete(oldest.value) +} interface McpServerDiscoveryFailure { serverId: string @@ -31,8 +72,13 @@ interface McpDiscoveryServers { /** Servers the request names: listed live and completely, however long that takes. */ complete: string[] /** - * Chat-only: servers available without being named. Served from the discovery - * cache when fresh, otherwise skipped after {@link BEST_EFFORT_DISCOVERY_WAIT_MS}. + * Chat-only: servers available without being named. At most + * {@link BEST_EFFORT_DISCOVERY_CONCURRENCY} are discovered at once, each served + * from the discovery cache when fresh and otherwise skipped after + * {@link BEST_EFFORT_DISCOVERY_WAIT_MS}; one still queued after + * {@link BEST_EFFORT_DISCOVERY_START_WINDOW_MS} is skipped unstarted. One that + * failed or outlived its wait is skipped for the same requester for + * {@link BEST_EFFORT_RETRY_AFTER_MS}, or until that late discovery succeeds. * One with no permitted operations contributes nothing and is not a failure. * Their discovery never touches a server's shared status or cooldown, and skips a * server the requester cannot authenticate to. @@ -92,23 +138,39 @@ export async function buildMcpToolSchemas( if (!tools.length) throw new Error(`No permitted MCP operations are available for ${serverId}`) return tools } + const startWindowEndsAt = Date.now() + BEST_EFFORT_DISCOVERY_START_WINDOW_MS /** Ignores the deadline's signal so a slow server's discovery still finishes and warms the cache. */ const discoverBestEffort = async (serverId: string) => { + const failureKey = `${workspaceId}:${serverId}:${userId}` + const recentFailure = recentBestEffortFailure(failureKey) + if (recentFailure) throw recentFailure + if (Date.now() >= startWindowEndsAt) throw new DeadlineExceededError() const discovery = discoverMcpServerToolsUseCase.execute({ principal: principal(), input: { workspaceId, serverId, bestEffort: true }, }) - const { tools } = await withinDeadline( - () => discovery, - Date.now() + BEST_EFFORT_DISCOVERY_WAIT_MS - ) - return tools + try { + const { tools } = await withinDeadline( + () => discovery, + Date.now() + BEST_EFFORT_DISCOVERY_WAIT_MS + ) + return tools + } catch (error) { + rememberBestEffortFailure(failureKey, toError(error)) + if (error instanceof DeadlineExceededError) + discovery.then( + () => recentBestEffortFailures.delete(failureKey), + (lateError: unknown) => rememberBestEffortFailure(failureKey, toError(lateError)) + ) + throw error + } } - const outcomes = await Promise.allSettled([ - ...complete.map(discoverComplete), - ...bestEffort.map(discoverBestEffort), + const [completeOutcomes, bestEffortOutcomes] = await Promise.all([ + Promise.allSettled(complete.map(discoverComplete)), + mapSettledWithConcurrency(bestEffort, BEST_EFFORT_DISCOVERY_CONCURRENCY, discoverBestEffort), ]) + const outcomes = [...completeOutcomes, ...bestEffortOutcomes] const seen = new Set() const discovery: McpDiscovery = { tools: [], failures: [] } From 038a465bfad890643c36e506b00e28f22e034ae2 Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Tue, 20 Oct 2026 21:21:04 -0700 Subject: [PATCH 2/2] fix(mothership): keep the best-effort memo in lru-cache and let only its own attempt settle it - Replace the hand-rolled TTL map with LRUCache (max 1000, ttl 60s), per the repo's caching rule. - A timed-out attempt's late outcome now settles the memo only while the entry is still that attempt's: an older attempt's late success no longer erases a newer attempt's failure, and its late failure no longer recreates a memo after a newer attempt succeeded. --- apps/sim/lib/mothership/mcp-tools.test.ts | 69 ++++++++++++++++++++++- apps/sim/lib/mothership/mcp-tools.ts | 53 ++++++++--------- 2 files changed, 94 insertions(+), 28 deletions(-) diff --git a/apps/sim/lib/mothership/mcp-tools.test.ts b/apps/sim/lib/mothership/mcp-tools.test.ts index 7fbdec49d34..0af8168e8d8 100644 --- a/apps/sim/lib/mothership/mcp-tools.test.ts +++ b/apps/sim/lib/mothership/mcp-tools.test.ts @@ -98,11 +98,26 @@ describe('mothership MCP best-effort discovery bounds', () => { const ids = (prefix: string, count: number) => Array.from({ length: count }, (_, index) => `${prefix}-${index + 1}`) + let clockStart = 0 + let clockOffset = 0 + /** + * `lru-cache` captures `performance` at load, so fake timers cannot move the memo's clock: spy it + * forward with the faked `Date` and timers. + */ + async function advanceClock(ms: number): Promise { + clockOffset += ms + vi.spyOn(performance, 'now').mockImplementation(() => clockStart + clockOffset) + await vi.advanceTimersByTimeAsync(ms) + } + beforeEach(() => { - vi.useFakeTimers() + vi.useFakeTimers({ toFake: ['setTimeout', 'clearTimeout', 'Date'] }) + clockStart = performance.now() + clockOffset = 0 discoverServerTools.mockReset() }) afterEach(() => { + vi.runOnlyPendingTimers() vi.useRealTimers() }) @@ -210,7 +225,7 @@ describe('mothership MCP best-effort discovery bounds', () => { await query('user-memo-b') expect(discoverServerTools).toHaveBeenCalledTimes(2) - vi.advanceTimersByTime(60_000) + await advanceClock(61_000) await query('user-memo-a') expect(discoverServerTools).toHaveBeenCalledTimes(3) }) @@ -237,4 +252,54 @@ describe('mothership MCP best-effort discovery bounds', () => { expect((await query()).tools.map((tool) => tool.service)).toEqual(['mcp:mcp-slow']) expect(discoverServerTools).toHaveBeenCalledTimes(2) }) + + it("keeps a newer attempt's failure when an older attempt succeeds late", async () => { + discoverServerTools + .mockImplementationOnce( + () => new Promise((resolve) => setTimeout(() => resolve(toolsOf('mcp-flaky')), 70_000)) + ) + .mockRejectedValueOnce(new Error('connect ECONNREFUSED')) + const query = () => + buildMcpToolSchemas('user-race-a', 'ws-race', { complete: [], bestEffort: ['mcp-flaky'] }) + + const first = query() + await advanceClock(3_000) + expect((await first).failures.map(({ serverId }) => serverId)).toEqual(['mcp-flaky']) + await advanceClock(61_000) + expect((await query()).failures.map(({ error }) => error.message)).toEqual([ + 'connect ECONNREFUSED', + ]) + + await advanceClock(6_000) + const third = query() + await flushMicrotasks(10) + expect(discoverServerTools).toHaveBeenCalledTimes(2) + expect((await third).failures.map(({ error }) => error.message)).toEqual([ + 'connect ECONNREFUSED', + ]) + }) + + it('does not remember a server whose older attempt fails late after a newer one succeeded', async () => { + discoverServerTools + .mockImplementationOnce( + () => + new Promise((_resolve, reject) => + setTimeout(() => reject(new Error('connect ETIMEDOUT')), 70_000) + ) + ) + .mockResolvedValueOnce(toolsOf('mcp-flaky')) + .mockResolvedValueOnce(toolsOf('mcp-flaky')) + const query = () => + buildMcpToolSchemas('user-race-b', 'ws-race', { complete: [], bestEffort: ['mcp-flaky'] }) + + const first = query() + await advanceClock(3_000) + expect((await first).failures.map(({ serverId }) => serverId)).toEqual(['mcp-flaky']) + await advanceClock(61_000) + expect((await query()).tools.map((tool) => tool.service)).toEqual(['mcp:mcp-flaky']) + + await advanceClock(6_000) + expect((await query()).tools.map((tool) => tool.service)).toEqual(['mcp:mcp-flaky']) + expect(discoverServerTools).toHaveBeenCalledTimes(3) + }) }) diff --git a/apps/sim/lib/mothership/mcp-tools.ts b/apps/sim/lib/mothership/mcp-tools.ts index 5e9aa9047b0..7997df93266 100644 --- a/apps/sim/lib/mothership/mcp-tools.ts +++ b/apps/sim/lib/mothership/mcp-tools.ts @@ -1,4 +1,5 @@ import { toError } from '@sim/utils/errors' +import { LRUCache } from 'lru-cache' import { mapSettledWithConcurrency } from '@/lib/core/utils/concurrency' import { DeadlineExceededError, withinDeadline } from '@/lib/core/utils/deadline' import { discoverMcpServerToolsAsExecutor } from '@/lib/internal/mcp/discover-tools' @@ -33,29 +34,21 @@ const BEST_EFFORT_DISCOVERY_START_WINDOW_MS = 3_000 const BEST_EFFORT_RETRY_AFTER_MS = 60_000 const MAX_RECENT_BEST_EFFORT_FAILURES = 1_000 +/** One attempt's failure; the entry's identity is that attempt's, so a later attempt's entry is never its to settle. */ +interface BestEffortFailure { + error: Error +} + /** * Best-effort servers that recently failed for a requester, keyed by - * `workspaceId:serverId:userId`, oldest first. It lives in this process because - * best-effort discovery must leave the server's shared status and cooldown to - * the discoveries someone asked for. + * `workspaceId:serverId:userId`. It lives in this process because best-effort + * discovery must leave the server's shared status and cooldown to the + * discoveries someone asked for. */ -const recentBestEffortFailures = new Map() - -function recentBestEffortFailure(key: string): Error | undefined { - const failure = recentBestEffortFailures.get(key) - if (!failure) return undefined - if (failure.retryAfter > Date.now()) return failure.error - recentBestEffortFailures.delete(key) - return undefined -} - -function rememberBestEffortFailure(key: string, error: Error): void { - recentBestEffortFailures.delete(key) - recentBestEffortFailures.set(key, { error, retryAfter: Date.now() + BEST_EFFORT_RETRY_AFTER_MS }) - if (recentBestEffortFailures.size <= MAX_RECENT_BEST_EFFORT_FAILURES) return - const oldest = recentBestEffortFailures.keys().next() - if (!oldest.done) recentBestEffortFailures.delete(oldest.value) -} +const recentBestEffortFailures = new LRUCache({ + max: MAX_RECENT_BEST_EFFORT_FAILURES, + ttl: BEST_EFFORT_RETRY_AFTER_MS, +}) interface McpServerDiscoveryFailure { serverId: string @@ -142,8 +135,8 @@ export async function buildMcpToolSchemas( /** Ignores the deadline's signal so a slow server's discovery still finishes and warms the cache. */ const discoverBestEffort = async (serverId: string) => { const failureKey = `${workspaceId}:${serverId}:${userId}` - const recentFailure = recentBestEffortFailure(failureKey) - if (recentFailure) throw recentFailure + const recentFailure = recentBestEffortFailures.get(failureKey) + if (recentFailure) throw recentFailure.error if (Date.now() >= startWindowEndsAt) throw new DeadlineExceededError() const discovery = discoverMcpServerToolsUseCase.execute({ principal: principal(), @@ -156,12 +149,20 @@ export async function buildMcpToolSchemas( ) return tools } catch (error) { - rememberBestEffortFailure(failureKey, toError(error)) - if (error instanceof DeadlineExceededError) + const failure: BestEffortFailure = { error: toError(error) } + recentBestEffortFailures.set(failureKey, failure) + if (error instanceof DeadlineExceededError) { + const isLatest = () => + recentBestEffortFailures.peek(failureKey, { allowStale: true }) === failure discovery.then( - () => recentBestEffortFailures.delete(failureKey), - (lateError: unknown) => rememberBestEffortFailure(failureKey, toError(lateError)) + () => { + if (isLatest()) recentBestEffortFailures.delete(failureKey) + }, + (lateError: unknown) => { + if (isLatest()) recentBestEffortFailures.set(failureKey, { error: toError(lateError) }) + } ) + } throw error } }