From 96dd3b677db741c148b88493d9b92cccff6d98b5 Mon Sep 17 00:00:00 2001 From: kugouming Date: Mon, 31 Aug 2026 22:53:19 +0800 Subject: [PATCH] =?UTF-8?q?test(pool):=20=E8=A1=A5=E9=BD=90=20transport=20?= =?UTF-8?q?error=20=E5=8D=B3=E6=97=B6=E5=A4=B1=E6=95=88=E5=8D=95=E6=B5=8B?= =?UTF-8?q?=E4=B8=8E=E5=90=8E=E7=AB=AF=E6=AD=BB=E4=BA=A1=E6=81=A2=E5=A4=8D?= =?UTF-8?q?=20E2E?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 基于 MERC-3 review 结论补齐测试缺口（MERC-4）： 单测（tests/unit/pool/transport-error-invalidation.test.ts）： - 覆盖 transport.on('error') → invalidateConnection 即时失效路径 （修复的卖点，此前完全没有测试覆盖） - 使用真实 BaseTransport 子类（TestTransport），走真实状态机 handleError → state ERROR + isConnected() 翻转 + emit('error')， 而非 mock isConnected()，可抓到 isConnectionHealthy 不再调用 isConnected() 或 handleError 不再翻 ERROR 的回归 - 验证：connectionFailed 携带连接 id、死连接释放 maxConnections 槽位、 下次 acquire 拿到不同 id 的新连接、幂等不重复失效 E2E（tests/integration/backend-death-recovery.test.ts）： - stdio 后端进程运行中退出（code 0）→ 池失效死连接 → 下次 tools/list 成功并经新连接返回相同 N 个工具（非 TRANSPORT_ERROR） - 验证单槽池在连接死亡后释放槽位、不再复用死连接 E2E（tests/integration/sse-reconnect-recovery.test.ts）： - SSE 后端重连耗尽 → transport ERROR → 池失效连接 → SSE 恢复可达后 下次 tools/list 经新连接成功；SSE 持续不可达时返回明确降级错误 （连接创建超时），而非吐回死连接的陈旧响应 验证：npm run typecheck/lint/format:check 通过；clean build 后 npm test 全绿（947/947，含 cli-mode 集成测试 9/9）。 --- .../backend-death-recovery.test.ts | 171 ++++++++++ .../integration/fixtures/mock-stdio-server.js | 140 +++++++++ .../sse-reconnect-recovery.test.ts | 293 ++++++++++++++++++ .../pool/transport-error-invalidation.test.ts | 223 +++++++++++++ 4 files changed, 827 insertions(+) create mode 100644 tests/integration/backend-death-recovery.test.ts create mode 100644 tests/integration/fixtures/mock-stdio-server.js create mode 100644 tests/integration/sse-reconnect-recovery.test.ts create mode 100644 tests/unit/pool/transport-error-invalidation.test.ts diff --git a/tests/integration/backend-death-recovery.test.ts b/tests/integration/backend-death-recovery.test.ts new file mode 100644 index 0000000..b21af5e --- /dev/null +++ b/tests/integration/backend-death-recovery.test.ts @@ -0,0 +1,171 @@ +/** + * Integration test: backend death does not permanently break tool discovery. + * + * Covers the Architect's stdio E2E gherkin (from the MERC-3 review): + * + * Feature: Backend death does not permanently break tool discovery + * Scenario: stdio backend process exits mid-run + * Given OneMCP server mode with a registered stdio backend, tools/list returns N tools + * When the backend child process exits (code 0) + * Then the next tools/list request succeeds (not TRANSPORT_ERROR) + * And returns the same N tools via a fresh connection + * + * This exercises the real StdioTransport + real ConnectionPool + a real child + * process (tests/integration/fixtures/mock-stdio-server.js). No transport is + * mocked, so it verifies the actual recovery chain the fix enables: + * process exit (code 0) → handleProcessExit → handleError → transport ERROR + * → pool's transport.on('error') listener → invalidateConnection + * → dead connection removed → next acquire creates a fresh connection + * → tools/list succeeds again. + * + * Without the fix (isConnectionHealthy ignoring transport state / no transport + * 'error' listener), the dead connection would be reused and the second + * tools/list would fail with TRANSPORT_ERROR / RESPONSE_STREAM_ENDED. + */ + +import { describe, it, expect, beforeEach, afterEach, vi } from 'vitest'; +import * as path from 'path'; +import { fileURLToPath } from 'url'; +import { ConnectionPool } from '../../src/pool/connection-pool.js'; +import type { ServiceDefinition, ConnectionPoolConfig } from '../../src/types/service.js'; +import type { JsonRpcMessage } from '../../src/types/jsonrpc.js'; + +const __dirname = path.dirname(fileURLToPath(import.meta.url)); +const FIXTURE = path.resolve(__dirname, 'fixtures', 'mock-stdio-server.js'); + +const POOL_CONFIG: ConnectionPoolConfig = { + maxConnections: 3, + idleTimeout: 60000, + connectionTimeout: 10000, +}; + +function makeStdioService(): ServiceDefinition { + return { + name: 'mock-stdio', + enabled: true, + tags: [], + transport: 'stdio', + command: 'node', + args: [FIXTURE], + connectionPool: POOL_CONFIG, + }; +} + +/** Minimal JSON-RPC client over a pooled connection: send a request and read + * the response with the matching id (skipping id-less notifications). */ +async function rpc( + transport: { send(m: JsonRpcMessage): Promise; receive(): AsyncIterator }, + id: string | number, + method: string, + params: unknown +): Promise<{ + result?: { tools?: Array<{ name: string }> }; + error?: { code: number; message: string }; +}> { + await transport.send({ jsonrpc: '2.0', id, method, params: params as never }); + const iterator = transport.receive(); + for (;;) { + const next = await iterator.next(); + if (next.done || !next.value) { + throw new Error('transport stream ended before a matching response'); + } + const msg = next.value as { id?: unknown; result?: unknown; error?: unknown }; + if (msg.id !== undefined && msg.id !== null && String(msg.id) === String(id)) { + return msg as { + result?: { tools?: Array<{ name: string }> }; + error?: { code: number; message: string }; + }; + } + } +} + +describe('Backend death does not permanently break tool discovery', () => { + describe('stdio backend process exits mid-run (code 0)', () => { + let pool: ConnectionPool; + + beforeEach(() => { + pool = new ConnectionPool(makeStdioService(), POOL_CONFIG); + // The pool re-emits transport errors on its own 'error' channel — attach a + // listener so Node's EventEmitter does not throw on an unhandled 'error'. + pool.on('error', () => {}); + }); + + afterEach(async () => { + if (pool) { + await pool.closeAll(); + } + }); + + it('recovers tools/list via a fresh connection after the backend exits', async () => { + // Given: a registered stdio backend, tools/list returns N tools. + const conn1 = await pool.acquire(); + const list1 = await rpc(conn1.transport, 'list-1', 'tools/list', {}); + expect(list1.error).toBeUndefined(); + expect(list1.result?.tools).toBeDefined(); + const tools1 = list1.result?.tools ?? []; + expect(tools1.length).toBe(3); + const names1 = tools1.map((t) => t.name); + + // When: the backend child process exits (code 0). + // The `exit` tool makes the fixture respond then process.exit(0). + const exitResp = await rpc(conn1.transport, 'exit-1', 'tools/call', { + name: 'exit', + arguments: {}, + }); + expect(exitResp.error).toBeUndefined(); + expect(exitResp.result).toBeDefined(); + + // The transport's process 'exit' (code 0) → handleError → transport ERROR + // → pool invalidates the dead connection. Wait for the slot to be freed. + await vi.waitFor(() => { + expect(pool.getStats().total).toBe(0); + }); + + // Then: the next tools/list request succeeds (not TRANSPORT_ERROR) and + // returns the same N tools via a fresh connection. + const conn2 = await pool.acquire(); + expect(conn2.id).not.toBe(conn1.id); // a fresh connection, not the dead one + + const list2 = await rpc(conn2.transport, 'list-2', 'tools/list', {}); + expect(list2.error).toBeUndefined(); + expect(list2.result?.tools).toBeDefined(); + const tools2 = list2.result?.tools ?? []; + expect(tools2.length).toBe(3); + expect(tools2.map((t) => t.name)).toEqual(names1); + + pool.release(conn2); + }, 30000); + + it('does not hand out the dead connection to a later acquire (slot freed)', async () => { + // Fill the single slot, kill the backend, and confirm the next acquire + // does not block/queue on the dead connection but creates a fresh one. + const singleSlotPool = new ConnectionPool(makeStdioService(), { + ...POOL_CONFIG, + maxConnections: 1, + }); + singleSlotPool.on('error', () => {}); + + try { + const conn1 = await singleSlotPool.acquire(); + expect(singleSlotPool.getStats().total).toBe(1); + + // Backend exits (code 0) → dead connection invalidated → slot freed. + await rpc(conn1.transport, 'exit-1', 'tools/call', { name: 'exit', arguments: {} }); + await vi.waitFor(() => { + expect(singleSlotPool.getStats().total).toBe(0); + }); + + // With the slot freed, a fresh connection is created instead of queueing. + const conn2 = await singleSlotPool.acquire(); + expect(conn2.id).not.toBe(conn1.id); + const list = await rpc(conn2.transport, 'list-1', 'tools/list', {}); + expect(list.error).toBeUndefined(); + expect(list.result?.tools?.length).toBe(3); + + singleSlotPool.release(conn2); + } finally { + await singleSlotPool.closeAll(); + } + }, 30000); + }); +}); diff --git a/tests/integration/fixtures/mock-stdio-server.js b/tests/integration/fixtures/mock-stdio-server.js new file mode 100644 index 0000000..8084f70 --- /dev/null +++ b/tests/integration/fixtures/mock-stdio-server.js @@ -0,0 +1,140 @@ +#!/usr/bin/env node +/** + * Mock MCP stdio backend for integration tests. + * + * Speaks NDJSON (one JSON-RPC message per line) on stdin/stdout. Exposes three + * tools (echo, add, exit) so a test can verify tools/list returns a stable N, + * and can trigger a clean "backend died mid-run" by calling the `exit` tool: + * the server responds, flushes stdout, then exits with code 0 — exactly the + * scenario the connection-pool fix must recover from. + * + * Used by tests/integration/backend-death-recovery.test.ts. + */ +'use strict'; + +const TOOLS = [ + { + name: 'echo', + description: 'Echo back the provided message', + inputSchema: { + type: 'object', + properties: { message: { type: 'string' } }, + required: ['message'], + }, + }, + { + name: 'add', + description: 'Add two numbers', + inputSchema: { + type: 'object', + properties: { a: { type: 'number' }, b: { type: 'number' } }, + required: ['a', 'b'], + }, + }, + { + name: 'exit', + description: 'Exit the backend with code 0 (simulates a backend that dies mid-run)', + inputSchema: { type: 'object', properties: {} }, + }, +]; + +function respond(msg) { + process.stdout.write(JSON.stringify(msg) + '\n'); +} + +function handle(req) { + // Notifications (no id) get no response. + if (req.id === undefined || req.id === null) { + return; + } + + if (req.method === 'initialize') { + respond({ + jsonrpc: '2.0', + id: req.id, + result: { + protocolVersion: '2024-11-05', + capabilities: {}, + serverInfo: { name: 'mock-stdio', version: '1.0.0' }, + }, + }); + return; + } + + if (req.method === 'tools/list') { + respond({ jsonrpc: '2.0', id: req.id, result: { tools: TOOLS } }); + return; + } + + if (req.method === 'tools/call') { + const name = req.params && req.params.name; + const args = (req.params && req.params.arguments) || {}; + + if (name === 'exit') { + // Respond, then exit code 0 AFTER the response is flushed to stdout so the + // client always receives the reply before the transport's process 'exit' + // fires — mirroring a backend that dies cleanly after serving a request. + const resp = { + jsonrpc: '2.0', + id: req.id, + result: { content: [{ type: 'text', text: 'exiting' }] }, + }; + process.stdout.write(JSON.stringify(resp) + '\n', () => process.exit(0)); + return; + } + + if (name === 'echo') { + respond({ + jsonrpc: '2.0', + id: req.id, + result: { content: [{ type: 'text', text: String(args.message ?? '') }] }, + }); + return; + } + + if (name === 'add') { + const sum = Number(args.a || 0) + Number(args.b || 0); + respond({ + jsonrpc: '2.0', + id: req.id, + result: { content: [{ type: 'text', text: String(sum) }] }, + }); + return; + } + + respond({ + jsonrpc: '2.0', + id: req.id, + error: { code: -32602, message: `Unknown tool: ${name}` }, + }); + return; + } + + respond({ + jsonrpc: '2.0', + id: req.id, + error: { code: -32601, message: `Unknown method: ${req.method}` }, + }); +} + +let buf = ''; +process.stdin.setEncoding('utf8'); +process.stdin.on('data', (chunk) => { + buf += chunk; + const lines = buf.split('\n'); + buf = lines.pop() || ''; + for (const line of lines) { + const trimmed = line.trim(); + if (!trimmed) continue; + let req; + try { + req = JSON.parse(trimmed); + } catch { + continue; + } + handle(req); + } +}); + +// Don't terminate just because stdin closes; let the pool drive the lifecycle. +process.stdin.on('end', () => {}); diff --git a/tests/integration/sse-reconnect-recovery.test.ts b/tests/integration/sse-reconnect-recovery.test.ts new file mode 100644 index 0000000..36c1acc --- /dev/null +++ b/tests/integration/sse-reconnect-recovery.test.ts @@ -0,0 +1,293 @@ +/** + * Integration test: SSE backend reconnect exhaustion → pool recovery. + * + * Covers the Architect's SSE E2E gherkin (from the MERC-3 review): + * + * Feature: Backend death does not permanently break tool discovery + * Scenario: SSE backend reconnect exhausts + * Given OneMCP server mode with a registered SSE backend, tools/list returns N tools + * When the backend SSE endpoint becomes unreachable past maxReconnectAttempts + * Then the pool marks the transport ERROR + * And the next tools/list either recovers once SSE is reachable again, + * or returns a clear degraded error (not a stale dead-connection response) + * + * HttpTransport and ConnectionPool are real; only the network primitives + * (eventsource, node-fetch) are mocked so the test can drive SSE connectivity + * deterministically. This verifies the real chain: + * SSE errors → handleSSEError reconnects → max attempts → handleError + * → transport state ERROR + 'error' emitted + * → pool's transport.on('error') listener → invalidateConnection + * → dead connection removed → next acquire creates a fresh transport + * → once SSE is reachable again, tools/list succeeds via the fresh connection. + */ + +import { describe, it, expect, beforeEach, afterEach, vi } from 'vitest'; +import { EventEmitter } from 'events'; +import EventSource from 'eventsource'; +import fetch from 'node-fetch'; +import { ConnectionPool } from '../../src/pool/connection-pool.js'; +import { TransportState } from '../../src/transport/base.js'; +import type { ServiceDefinition, ConnectionPoolConfig } from '../../src/types/service.js'; +import type { JsonRpcMessage } from '../../src/types/jsonrpc.js'; + +vi.mock('eventsource'); +vi.mock('node-fetch'); + +const POOL_CONFIG: ConnectionPoolConfig = { + maxConnections: 3, + idleTimeout: 60000, + connectionTimeout: 10000, +}; + +/** + * A controllable EventSource. HttpTransport uses addEventListener('endpoint'), + * onmessage, onopen, onerror and close(). This mock is a real EventEmitter so + * the transport's listeners really register, and connectivity is driven by + * `sseReachable`: when reachable, a new EventSource completes the MCP SSE + * handshake (endpoint + an initialize response); when unreachable, it stays + * silent so the transport's reconnect logic must exhaust. + */ +class MockEventSource extends EventEmitter { + public onmessage: ((e: { data: string }) => void) | null = null; + public onopen: ((e: Event) => void) | null = null; + public onerror: ((e: Event) => void) | null = null; + public close = vi.fn(); + // HttpTransport calls addEventListener('endpoint', handler); alias to on(). + public addEventListener = this.on.bind(this); + + constructor(private readonly reachable: boolean) { + super(); + // Drive the SSE handshake / unreachability on a microtask so HttpTransport + // has assigned onmessage/onopen/onerror first. + queueMicrotask(() => { + if (this.reachable) { + // Standard MCP SSE handshake: server sends the 'endpoint' event. + this.emit('endpoint', { data: '/messages' }); + // Then push an initialize response so the pool's MCP init completes. + queueMicrotask(() => { + if (this.onmessage) { + this.onmessage({ + data: JSON.stringify({ jsonrpc: '2.0', id: 'init', result: {} }), + }); + } + }); + } + // When unreachable, do nothing — onerror is driven by the test via + // simulateError() to step the transport's reconnect logic. + }); + } + + public simulateMessage(data: string): void { + if (this.onmessage) this.onmessage({ data }); + } + + public simulateError(): void { + if (this.onerror) this.onerror({} as Event); + } +} + +describe('Backend death does not permanently break tool discovery', () => { + describe('SSE backend reconnect exhausts', () => { + let pool: ConnectionPool; + let createdEventSources: MockEventSource[]; + let sseReachable: boolean; + + beforeEach(() => { + // Fake only timer APIs so the reconnect setTimeout chain is deterministic + // while microtasks (queueMicrotask / promises) still run normally. + vi.useFakeTimers({ toFake: ['setTimeout', 'setInterval', 'Date'] }); + vi.spyOn(console, 'warn').mockImplementation(() => {}); + vi.spyOn(console, 'error').mockImplementation(() => {}); + + createdEventSources = []; + sseReachable = true; + + vi.mocked(EventSource).mockImplementation(() => { + const es = new MockEventSource(sseReachable); + createdEventSources.push(es); + return es as unknown as EventSource; + }); + + vi.mocked(fetch).mockResolvedValue({ + ok: true, + status: 200, + statusText: 'OK', + headers: { get: () => null }, + text: () => Promise.resolve(''), + } as never); + + const service: ServiceDefinition = { + name: 'mock-sse', + enabled: true, + tags: [], + transport: 'sse', + url: 'http://localhost:65535/sse', + connectionPool: POOL_CONFIG, + }; + pool = new ConnectionPool(service, POOL_CONFIG); + pool.on('error', () => {}); + }); + + afterEach(async () => { + if (pool) { + await pool.closeAll(); + } + vi.useRealTimers(); + vi.restoreAllMocks(); + }); + + const currentEventSource = (): MockEventSource => { + const es = createdEventSources[createdEventSources.length - 1]; + if (!es) throw new Error('no EventSource created yet'); + return es; + }; + + /** Send a JSON-RPC request over the connection and read the matching reply. */ + async function rpc( + transport: { + send(m: JsonRpcMessage): Promise; + receive(): AsyncIterator; + }, + id: string, + method: string, + params: unknown + ): Promise<{ + result?: { tools?: Array<{ name: string }> }; + error?: { code: number; message: string }; + }> { + await transport.send({ jsonrpc: '2.0', id, method, params: params as never }); + const iterator = transport.receive(); + for (;;) { + const next = await iterator.next(); + if (next.done || !next.value) { + throw new Error('transport stream ended before a matching response'); + } + const msg = next.value as { id?: unknown; result?: unknown; error?: unknown }; + if (msg.id !== undefined && msg.id !== null && String(msg.id) === String(id)) { + return msg as { + result?: { tools?: Array<{ name: string }> }; + error?: { code: number; message: string }; + }; + } + } + } + + const TOOLS_LIST_RESPONSE = JSON.stringify({ + jsonrpc: '2.0', + id: 'list-1', + result: { + tools: [ + { name: 'echo', description: 'echo', inputSchema: { type: 'object', properties: {} } }, + { name: 'add', description: 'add', inputSchema: { type: 'object', properties: {} } }, + ], + }, + }); + + it('marks the transport ERROR and recovers tools/list via a fresh connection once SSE is reachable', async () => { + // Given: a registered SSE backend, tools/list returns N tools. + const conn1 = await pool.acquire(); + // The reachable EventSource auto-completed the init handshake; push the + // tools/list reply and read it. + currentEventSource().simulateMessage(TOOLS_LIST_RESPONSE); + const list1 = await rpc(conn1.transport, 'list-1', 'tools/list', {}); + expect(list1.error).toBeUndefined(); + expect(list1.result?.tools?.length).toBe(2); + pool.release(conn1); // idle in the pool + + // Capture the moment the transport reaches ERROR (handleError sets state + // ERROR and emits 'error' before the pool's close() moves it to CLOSED). + const transportErrorSpy = vi.fn(); + conn1.transport.on('error', transportErrorSpy); + + const connectionFailedSpy = vi.fn(); + pool.on('connectionFailed', connectionFailedSpy); + + // When: the SSE endpoint becomes unreachable past maxReconnectAttempts. + sseReachable = false; // any reconnect attempt stays unreachable + const maxAttempts = 3; // HttpTransport default maxReconnectAttempts + // Kick off the first error on the live connection; each reconnect creates + // a new (unreachable) EventSource whose error advances the counter. + currentEventSource().simulateError(); + for (let i = 0; i < maxAttempts; i++) { + // Fire the scheduled reconnect (exponential backoff: 1s, 2s, 4s). + await vi.advanceTimersByTimeAsync(2 ** i * 1000 + 1); + // The reconnect created a new unreachable EventSource; step its error. + currentEventSource().simulateError(); + } + + // Then: the transport reached ERROR and the pool invalidated the connection. + expect(transportErrorSpy).toHaveBeenCalled(); + expect( + (conn1.transport as unknown as { getState: () => TransportState }).getState() + ).not.toBe(TransportState.CONNECTED); + expect((conn1.transport as { isConnected: () => boolean }).isConnected()).toBe(false); + + await vi.waitFor(() => { + expect(connectionFailedSpy).toHaveBeenCalledWith(conn1.id, expect.any(Error)); + }); + expect(pool.getStats().total).toBe(0); // dead connection no longer occupies a slot + + // And: once SSE is reachable again, the next tools/list recovers via a + // fresh connection (not a stale dead-connection response). + sseReachable = true; + const conn2 = await pool.acquire(); + expect(conn2.id).not.toBe(conn1.id); // fresh connection + + currentEventSource().simulateMessage( + JSON.stringify({ + jsonrpc: '2.0', + id: 'list-2', + result: { + tools: [ + { + name: 'echo', + description: 'echo', + inputSchema: { type: 'object', properties: {} }, + }, + { name: 'add', description: 'add', inputSchema: { type: 'object', properties: {} } }, + ], + }, + }) + ); + const list2 = await rpc(conn2.transport, 'list-2', 'tools/list', {}); + expect(list2.error).toBeUndefined(); + expect(list2.result?.tools?.map((t) => t.name)).toEqual(['echo', 'add']); + + pool.release(conn2); + }, 30000); + + it('does not hand out the stale dead SSE connection (returns a clear error while unreachable)', async () => { + // While SSE stays unreachable, the next acquire cannot reuse the dead + // connection — it must attempt a fresh transport and fail clearly (not + // return a stale dead-connection response). + const conn1 = await pool.acquire(); + currentEventSource().simulateMessage(TOOLS_LIST_RESPONSE); + const list1 = await rpc(conn1.transport, 'list-1', 'tools/list', {}); + expect(list1.result?.tools?.length).toBe(2); + pool.release(conn1); + + sseReachable = false; + currentEventSource().simulateError(); + const maxAttempts = 3; + for (let i = 0; i < maxAttempts; i++) { + await vi.advanceTimersByTimeAsync(2 ** i * 1000 + 1); + currentEventSource().simulateError(); + } + await vi.waitFor(() => expect(pool.getStats().total).toBe(0)); + + // SSE is still unreachable: a fresh acquire attempts a new transport that + // cannot connect, so the pool rejects with a clear error (connection + // creation timeout) rather than handing back the dead connection. + sseReachable = false; + const acquirePromise = pool.acquire(); + acquirePromise.catch(() => {}); // avoid unhandled rejection during the wait + // The unreachable EventSource never completes the SSE handshake, so + // createTransport's bounded wait rejects after connectionTimeout. + await vi.advanceTimersByTimeAsync(POOL_CONFIG.connectionTimeout + 1000); + await expect(acquirePromise).rejects.toThrow(); + + // The dead connection is not resurrected: the pool is still empty. + expect(pool.getStats().total).toBe(0); + }, 30000); + }); +}); diff --git a/tests/unit/pool/transport-error-invalidation.test.ts b/tests/unit/pool/transport-error-invalidation.test.ts new file mode 100644 index 0000000..881c499 --- /dev/null +++ b/tests/unit/pool/transport-error-invalidation.test.ts @@ -0,0 +1,223 @@ +/** + * Unit tests: transport error → pool connection invalidation (the full chain) + * + * These tests address the gap the Architect's review flagged: the existing pool + * suites mock isConnected() directly (returning false), so they only verify the + * pool's reaction to an already-unhealthy connection — not the real chain + * transport enters ERROR → isConnected() flips → pool invalidates → fresh acquire. + * + * They also cover the fix's "instant invalidation" selling point — + * `transport.on('error') → invalidateConnection` — which previously had no test + * coverage at all. + * + * To exercise the real state machine, the pool is wired to a TestTransport that + * extends the real BaseTransport (a real EventEmitter). TestTransport uses + * BaseTransport's actual handleError(): it flips state to ERROR and emits 'error' + * exactly like StdioTransport.handleProcessExit / HttpTransport.handleSSEError do, + * so isConnected() reflects a real state transition, not a stubbed return value. + */ + +import { describe, it, expect, beforeEach, afterEach, vi } from 'vitest'; +import { BaseTransport } from '../../../src/transport/base.js'; +import type { JsonRpcMessage } from '../../../src/types/jsonrpc.js'; +import type { + TransportType, + ServiceDefinition, + ConnectionPoolConfig, +} from '../../../src/types/service.js'; +import { ConnectionPool } from '../../../src/pool/connection-pool.js'; +import { StdioTransport } from '../../../src/transport/stdio.js'; + +/** + * Real EventEmitter-backed transport using BaseTransport's actual state machine. + * + * - on()/emit() come from EventEmitter, so the pool's `transport.on('error', ...)` + * listener is really registered (not a vi.fn() no-op). + * - isConnected() comes from BaseTransport and flips to false when handleError() + * moves state to ERROR — the same path StdioTransport/HttpTransport use. + * - simulateError() is the test hook for that real path. + */ +class TestTransport extends BaseTransport { + // Present so the stdio branch of isConnectionHealthy can inspect it. + public process = { killed: false, exitCode: null as number | null }; + public closeCalls = 0; + + constructor() { + super(); + // Match StdioTransport: once spawned, the transport is connected. + this.setConnected(); + } + + public getType(): TransportType { + return 'stdio'; + } + + protected async doSend(_message: JsonRpcMessage): Promise { + // no-op — the pool only sends initialize + initialized notifications + } + + protected async *doReceive(): AsyncIterator { + // Yield a valid initialize response so the pool's MCP handshake completes. + yield { jsonrpc: '2.0', id: 1, result: {} } as JsonRpcMessage; + } + + protected async doClose(): Promise { + this.closeCalls++; + } + + /** + * Trigger the real error path, identical to StdioTransport.handleProcessExit + * and HttpTransport.handleSSEError: flip state to ERROR and emit 'error'. + */ + public simulateError(error: Error = new Error('simulated transport error')): void { + this.handleError(error); + } +} + +// The pool constructs transports via `new StdioTransport(config)`. Replace that +// constructor with one that returns a TestTransport, so the pool exercises the +// real BaseTransport state machine instead of a vi.fn()-stubbed isConnected(). +vi.mock('../../../src/transport/stdio.js', () => ({ + StdioTransport: vi.fn(), +})); + +describe('Transport error invalidates pool connection', () => { + let pool: ConnectionPool; + let service: ServiceDefinition; + let poolConfig: ConnectionPoolConfig; + + beforeEach(() => { + vi.mocked(StdioTransport).mockImplementation(function (this: unknown) { + return new TestTransport() as unknown as StdioTransport; + } as never); + + service = { + name: 'test-service', + enabled: true, + tags: [], + transport: 'stdio', + command: 'test-command', + connectionPool: { + maxConnections: 3, + idleTimeout: 60000, + connectionTimeout: 5000, + }, + }; + poolConfig = { + maxConnections: 3, + idleTimeout: 60000, + connectionTimeout: 5000, + }; + pool = new ConnectionPool(service, poolConfig); + // The pool re-emits transport errors on its own 'error' channel. Without a + // listener Node's EventEmitter would throw on an unhandled 'error' event. + pool.on('error', () => {}); + }); + + afterEach(async () => { + if (pool) { + await pool.closeAll(); + } + }); + + it('a pooled transport error emits connectionFailed with that connection id and frees the slot', async () => { + const conn = await pool.acquire(); + pool.release(conn); // idle in the pool + + const failedSpy = vi.fn(); + pool.on('connectionFailed', failedSpy); + + // When the transport emits a real error (process exit / SSE exhausted / HTTP + // failure), the pool must invalidate the connection immediately. + (conn.transport as TestTransport).simulateError(); + + await vi.waitFor(() => { + expect(failedSpy).toHaveBeenCalledWith(conn.id, expect.any(Error)); + }); + + // The dead connection no longer occupies a maxConnections slot. + expect(pool.getStats().total).toBe(0); + + // The next acquire returns a fresh connection with a different id. + const conn2 = await pool.acquire(); + expect(conn2.id).not.toBe(conn.id); + expect(conn2.transport.isConnected()).toBe(true); + }); + + it('invalidates a busy connection that errors mid-use (not just idle ones)', async () => { + const conn = await pool.acquire(); // busy, never released + + const failedSpy = vi.fn(); + pool.on('connectionFailed', failedSpy); + + (conn.transport as TestTransport).simulateError(); + + await vi.waitFor(() => { + expect(failedSpy).toHaveBeenCalledWith(conn.id, expect.any(Error)); + }); + expect(pool.getStats().total).toBe(0); + + const conn2 = await pool.acquire(); + expect(conn2.id).not.toBe(conn.id); + }); + + it('is idempotent: a second error on the same connection does not double-invalidate', async () => { + const conn = await pool.acquire(); + pool.release(conn); + + const failedSpy = vi.fn(); + pool.on('connectionFailed', failedSpy); + + (conn.transport as TestTransport).simulateError(); + await vi.waitFor(() => expect(failedSpy).toHaveBeenCalledTimes(1)); + + // A second error arrives (e.g. the 'exit' event after the stream 'error'). + (conn.transport as TestTransport).simulateError(); + await new Promise((resolve) => setImmediate(resolve)); + + // connectionFailed must fire exactly once — invalidateConnection is idempotent. + expect(failedSpy).toHaveBeenCalledTimes(1); + expect(pool.getStats().total).toBe(0); + }); + + it('does not reuse a connection whose transport reached ERROR via the real state machine', async () => { + const conn = await pool.acquire(); + pool.release(conn); // idle in the pool + + // The transport really enters ERROR — isConnected() flips for real, it is not + // stubbed. This is the regression guard: if isConnectionHealthy stops calling + // isConnected(), or handleError stops flipping state, this test fails. + (conn.transport as TestTransport).simulateError(); + await new Promise((resolve) => setImmediate(resolve)); + + expect((conn.transport as TestTransport).isConnected()).toBe(false); + expect(pool.isConnectionHealthy(conn)).toBe(false); + + // The dead idle connection must not be handed out — a fresh one is created. + const conn2 = await pool.acquire(); + expect(conn2.id).not.toBe(conn.id); + expect(pool.isConnectionHealthy(conn2)).toBe(true); + expect(pool.getStats().total).toBe(1); + }); + + it('freeing the slot prevents pool exhaustion when a connection dies at max capacity', async () => { + // A pool with a single slot: if the dead connection kept occupying it, the + // next acquire would queue/timeout instead of getting a fresh connection. + const singleSlotPool = new ConnectionPool(service, { ...poolConfig, maxConnections: 1 }); + singleSlotPool.on('error', () => {}); + + const conn = await singleSlotPool.acquire(); + expect(singleSlotPool.getStats().total).toBe(1); + singleSlotPool.release(conn); + + (conn.transport as TestTransport).simulateError(); + await vi.waitFor(() => expect(singleSlotPool.getStats().total).toBe(0)); + + // With the slot freed, acquire returns a fresh connection instead of queueing. + const conn2 = await singleSlotPool.acquire(); + expect(conn2.id).not.toBe(conn.id); + expect(singleSlotPool.getStats().total).toBe(1); + + await singleSlotPool.closeAll(); + }); +});