diff --git a/apps/sim/lib/mcp/connection-pool.test.ts b/apps/sim/lib/mcp/connection-pool.test.ts index 4df677f13e0..9359eceb92e 100644 --- a/apps/sim/lib/mcp/connection-pool.test.ts +++ b/apps/sim/lib/mcp/connection-pool.test.ts @@ -83,6 +83,50 @@ describe('McpConnectionPool', () => { expect(client.disconnect).not.toHaveBeenCalled() }) + it('keeps the connection after a single timeout release', async () => { + const client = makeFakeClient() + const create = vi.fn(async () => client) + + const lease = await pool.acquire(params('s1:w1:u1', create)) + await lease.release(false, true) + await borrow(pool, params('s1:w1:u1', create)) + + expect(create).toHaveBeenCalledTimes(1) + expect(client.disconnect).not.toHaveBeenCalled() + }) + + it('retires the connection after consecutive timeouts (circuit breaker)', async () => { + const client = makeFakeClient() + const replacement = makeFakeClient() + const create = vi.fn<() => Promise>() + create.mockResolvedValueOnce(client) + create.mockResolvedValue(replacement) + + const l1 = await pool.acquire(params('s1:w1:u1', create)) + await l1.release(false, true) + const l2 = await pool.acquire(params('s1:w1:u1', create)) + await l2.release(false, true) + + expect(client.disconnect).toHaveBeenCalledTimes(1) + const l3 = await pool.acquire(params('s1:w1:u1', create)) + expect(l3.client).toBe(replacement) + await l3.release() + }) + + it('resets the timeout count on a healthy release', async () => { + const client = makeFakeClient() + const create = vi.fn(async () => client) + + const l1 = await pool.acquire(params('s1:w1:u1', create)) + await l1.release(false, true) + await borrow(pool, params('s1:w1:u1', create)) // healthy — resets the count + const l3 = await pool.acquire(params('s1:w1:u1', create)) + await l3.release(false, true) // first of a new streak, not the second strike + + expect(client.disconnect).not.toHaveBeenCalled() + expect(create).toHaveBeenCalledTimes(1) + }) + it('dedups concurrent creates into a single connect (single-flight)', async () => { let resolveCreate: ((client: McpClient) => void) | undefined const created = new Promise((resolve) => { diff --git a/apps/sim/lib/mcp/connection-pool.ts b/apps/sim/lib/mcp/connection-pool.ts index 47e162668ca..8eda72cbd3e 100644 --- a/apps/sim/lib/mcp/connection-pool.ts +++ b/apps/sim/lib/mcp/connection-pool.ts @@ -32,6 +32,14 @@ const logger = createLogger('McpConnectionPool') const MAX_POOL_SIZE = 100 /** Max lifetime; on expiry the pinned SSRF `resolvedIP` re-resolves. */ const MAX_CONNECTION_AGE_MS = 10 * 60 * 1000 +/** + * Circuit breaker: retire a connection after this many CONSECUTIVE request + * timeouts. One timeout is a slow request and keeps the session (retiring on + * every timeout causes connect/stall/reconnect churn); repeated timeouts with + * no success in between indicate a half-open transport the liveness ping + * hasn't caught yet. A healthy release resets the count. + */ +const MAX_CONSECUTIVE_TIMEOUTS = 2 const LIVENESS_TTL_MS = 60 * 1000 const LIVENESS_PING_TIMEOUT_MS = 5 * 1000 const IDLE_TIMEOUT_MS = 5 * 60 * 1000 @@ -46,10 +54,14 @@ export interface AcquireParams { create: () => Promise } -/** A borrowed connection. `release(poison)` must be called exactly once; pass `poison: true` to retire on failure. */ +/** + * A borrowed connection. `release(poison, sawTimeout)` must be called exactly once: + * `poison: true` retires immediately (dead-connection error); `sawTimeout: true` + * counts toward the consecutive-timeout circuit breaker without retiring on its own. + */ export interface ConnectionLease { client: McpClient - release: (poison?: boolean) => Promise + release: (poison?: boolean, sawTimeout?: boolean) => Promise } interface PoolEntry { @@ -60,6 +72,7 @@ interface PoolEntry { lastActivityAt: number lastLivenessCheckAt: number borrowers: number + consecutiveTimeouts: number retired: boolean closing: boolean } @@ -95,7 +108,10 @@ export class McpConnectionPool { while (entry.closing) entry = await this.resolveEntry(params) entry.borrowers++ entry.lastActivityAt = Date.now() - return { client: entry.client, release: (poison) => this.release(entry, poison ?? false) } + return { + client: entry.client, + release: (poison, sawTimeout) => this.release(entry, poison ?? false, sawTimeout ?? false), + } } private async resolveEntry(params: AcquireParams): Promise { @@ -162,6 +178,7 @@ export class McpConnectionPool { lastActivityAt: now, lastLivenessCheckAt: now, borrowers: 0, + consecutiveTimeouts: 0, retired: false, closing: false, } @@ -194,9 +211,17 @@ export class McpConnectionPool { return record.promise } - private async release(entry: PoolEntry, poison: boolean): Promise { + private async release(entry: PoolEntry, poison: boolean, sawTimeout: boolean): Promise { entry.borrowers = Math.max(0, entry.borrowers - 1) entry.lastActivityAt = Date.now() + if (sawTimeout) { + entry.consecutiveTimeouts++ + if (entry.consecutiveTimeouts >= MAX_CONSECUTIVE_TIMEOUTS) { + this.retire(entry, `${entry.consecutiveTimeouts} consecutive request timeouts`) + } + } else if (!poison) { + entry.consecutiveTimeouts = 0 + } if (poison) this.retire(entry, 'poisoned by failed operation') await this.closeIfIdle(entry) } diff --git a/apps/sim/lib/mcp/service-pool.test.ts b/apps/sim/lib/mcp/service-pool.test.ts index f2de2cab54e..be7fb6de847 100644 --- a/apps/sim/lib/mcp/service-pool.test.ts +++ b/apps/sim/lib/mcp/service-pool.test.ts @@ -134,7 +134,7 @@ describe('McpService connection reuse wiring', () => { expect.objectContaining({ key: `server-1:${WORKSPACE_ID}:${USER_ID}`, serverId: 'server-1' }) ) expect(mockCallTool).toHaveBeenCalledTimes(1) - expect(mockRelease).toHaveBeenCalledWith(false) + expect(mockRelease).toHaveBeenCalledWith(false, false) expect(poolClient.disconnect).not.toHaveBeenCalled() // A pool hit must not re-resolve env vars (acquire never invoked `create`). expect(mockResolveEnvVars).not.toHaveBeenCalled() @@ -158,7 +158,7 @@ describe('McpService connection reuse wiring', () => { mcpService.executeTool(USER_ID, 'server-1', { name: 'do', arguments: {} }, WORKSPACE_ID) ).rejects.toThrow() - expect(mockRelease).toHaveBeenCalledWith(true) + expect(mockRelease).toHaveBeenCalledWith(true, false) }) it('keeps the pooled connection warm on a benign (non-connection) tool error', async () => { @@ -168,20 +168,37 @@ describe('McpService connection reuse wiring', () => { mcpService.executeTool(USER_ID, 'server-1', { name: 'do', arguments: {} }, WORKSPACE_ID) ).rejects.toThrow() - expect(mockRelease).toHaveBeenCalledWith(false) + expect(mockRelease).toHaveBeenCalledWith(false, false) }) it('keeps the pooled connection warm on a request timeout (does not retire the session)', async () => { // A streamable-HTTP request timeout aborts only that request's stream; the session stays // healthy for the next request, so a timeout must NOT poison the lease (matches every - // production MCP client and avoids a connect/stall/reconnect churn loop). + // production MCP client and avoids a connect/stall/reconnect churn loop). It DOES report + // sawTimeout so the pool's consecutive-timeout circuit breaker can retire a half-open + // transport after repeated strikes. mockCallTool.mockRejectedValue(new Error('Request timed out')) await expect( mcpService.executeTool(USER_ID, 'server-1', { name: 'do', arguments: {} }, WORKSPACE_ID) ).rejects.toThrow() - expect(mockRelease).not.toHaveBeenCalledWith(true) + expect(mockRelease).toHaveBeenCalledWith(false, true) + }) + + it('classifies an AbortSignal.timeout-shaped TimeoutError as a timeout for the breaker', async () => { + // DOMException name 'TimeoutError' with a message that lacks "timed out". + mockCallTool.mockRejectedValue( + Object.assign(new Error('The operation was aborted due to timeout'), { + name: 'TimeoutError', + }) + ) + + await expect( + mcpService.executeTool(USER_ID, 'server-1', { name: 'do', arguments: {} }, WORKSPACE_ID) + ).rejects.toThrow() + + expect(mockRelease).toHaveBeenCalledWith(false, true) }) it('poisons the lease on an auth failure so a rotated credential is re-resolved', async () => { @@ -191,7 +208,7 @@ describe('McpService connection reuse wiring', () => { mcpService.executeTool(USER_ID, 'server-1', { name: 'do', arguments: {} }, WORKSPACE_ID) ).rejects.toThrow() - expect(mockRelease).toHaveBeenCalledWith(true) + expect(mockRelease).toHaveBeenCalledWith(true, false) }) it('retries and recovers when a rotated credential causes a one-off auth failure', async () => { @@ -209,7 +226,7 @@ describe('McpService connection reuse wiring', () => { expect(result).toEqual({ content: [] }) expect(mockCallTool).toHaveBeenCalledTimes(2) // First attempt poisoned the stale lease; the retry re-acquired a fresh one. - expect(mockRelease).toHaveBeenCalledWith(true) + expect(mockRelease).toHaveBeenCalledWith(true, false) expect(mockAcquire).toHaveBeenCalledTimes(2) }) }) diff --git a/apps/sim/lib/mcp/service.ts b/apps/sim/lib/mcp/service.ts index ac595424e8e..d4ef715b8e2 100644 --- a/apps/sim/lib/mcp/service.ts +++ b/apps/sim/lib/mcp/service.ts @@ -93,6 +93,12 @@ function isTimeoutError(error: unknown): boolean { if (error instanceof McpError && error.code === ErrorCode.RequestTimeout) { return true } + // AbortSignal.timeout / undici surface a DOMException named TimeoutError whose + // message ("The operation was aborted due to timeout") lacks "timed out". + const e = error as { name?: string; cause?: { name?: string } } | null + if (e?.name === 'TimeoutError' || e?.cause?.name === 'TimeoutError') { + return true + } return getErrorMessage(error, '').toLowerCase().includes('timed out') } @@ -425,13 +431,17 @@ class McpService { create, }) let poison = false + let sawTimeout = false try { return await fn(lease.client) } catch (error) { poison = isDeadConnectionError(error) + // A lone timeout keeps the session; the pool's circuit breaker retires it + // after consecutive timeouts with no healthy request in between. + sawTimeout = isTimeoutError(error) throw error } finally { - await lease.release(poison) + await lease.release(poison, sawTimeout) } }