fix(mcp): retire pooled connections after consecutive request timeouts (#5821)

* fix(mcp): retire pooled connections after consecutive request timeouts

Circuit breaker for the half-open-transport gap left by #5817: a lone timeout
still keeps the session warm (retiring on every timeout caused the
connect/stall/reconnect churn), but two consecutive timeouts with no healthy
request in between retire the connection so a genuinely half-open transport
can't serve repeated 30s failures until the liveness ping catches it. A healthy
release resets the count. Matches the LibreChat pattern of counting
connection-level failures rather than reacting to one.

* fix(mcp): classify TimeoutError-named aborts as timeouts for the circuit breaker

AbortSignal.timeout / undici surface a DOMException named TimeoutError whose
message lacks 'timed out'; without this the breaker treated it as a healthy
release and reset the streak. Adds name-based detection (incl. cause) + test.

Author: waleedlatif1

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<McpClient>>()
+    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<McpClient>((resolve) => {

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<McpClient>
 }
 
-/** 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<void>
+  release: (poison?: boolean, sawTimeout?: boolean) => Promise<void>
 }
 
 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<PoolEntry> {
@@ -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<void> {
+  private async release(entry: PoolEntry, poison: boolean, sawTimeout: boolean): Promise<void> {
     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)
   }

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)
   })
 })

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)
       }
     }