[MERC-3] 修复连接池健康判定，transport error 即时失效连接

src/pool/connection-pool.ts

@@ -287,6 +287,13 @@ export class ConnectionPool extends EventEmitter {
       return false;
     }
 
+    // Transport must be in CONNECTED state. A transport that reached ERROR
+    // (SSE reconnect exhausted, HTTP request failure, process exit) must not
+    // be reused — otherwise the pool keeps handing out a dead connection.
+    if (!connection.transport.isConnected()) {
+      return false;
+    }
+
     // For stdio transport, check if the process is still running
     if (connection.transport.getType() === 'stdio') {
       const stdioTransport = connection.transport as {
@@ -396,17 +403,68 @@ export class ConnectionPool extends EventEmitter {
   /**
    * Find an idle connection in the pool
    *
-   * @returns Idle connection or undefined if none available
+   * Only healthy idle connections are returned. If an idle connection is no
+   * longer healthy (e.g. its transport reached ERROR), it is removed from the
+   * pool so the next acquire creates a fresh connection instead of reusing a
+   * dead one or letting it occupy a slot up to maxConnections.
+   *
+   * @returns Healthy idle connection or undefined if none available
    */
   private findIdleConnection(): Connection | undefined {
+    let unhealthy: Connection | undefined;
+
     for (const connection of this.connections.values()) {
-      if (isIdle(connection)) {
+      if (!isIdle(connection)) {
+        continue;
+      }
+
+      if (this.isConnectionHealthy(connection)) {
+        // If we passed an unhealthy idle connection on the way, remove it first
+        if (unhealthy) {
+          void this.invalidateConnection(unhealthy, new Error('Connection is unhealthy'));
+        }
         return connection;
       }
+
+      if (!unhealthy) {
+        unhealthy = connection;
+      }
     }
+
+    if (unhealthy) {
+      void this.invalidateConnection(unhealthy, new Error('Connection is unhealthy'));
+    }
+
     return undefined;
   }
 
+  /**
+   * Remove a connection from the pool.
+   *
+   * Idempotent: drops the connection from the pool immediately (so a dead
+   * connection does not occupy a maxConnections slot), closes its transport,
+   * and awakes any queued requests so they get a fresh connection.
+   */
+  private async invalidateConnection(connection: Connection, error: Error): Promise<void> {
+    if (!this.connections.has(connection.id)) {
+      return;
+    }
+
+    // Drop from the pool synchronously before awaiting the transport close so
+    // concurrent health checks / request failures cannot double-remove it.
+    this.connections.delete(connection.id);
+    this.emit('connectionFailed', connection.id, error);
+
+    try {
+      await connection.transport.close();
+      this.emit('connectionClosed', connection.id);
+    } catch (closeError) {
+      this.emit('error', closeError);
+    }
+
+    this.processQueue();
+  }
+
   /**
    * Create a new connection
    *
@@ -468,11 +526,15 @@ export class ConnectionPool extends EventEmitter {
       const connection = createConnection(id, transport);
 
       // Attach the listener before initialization so an early process exit does not
-      // surface as an unhandled EventEmitter error.
+      // surface as an unhandled EventEmitter error. A transport error (SSE reconnect
+      // exhausted, HTTP request failure, process exit) makes the connection unusable:
+      // invalidate it in the pool so it is not reused or counted against maxConnections.
       transport.on('error', (error: unknown) => {
         const errorMessage = error instanceof Error ? error.message : String(error);
         log.warn(`[${this.service.name}] Transport error: ${errorMessage}`);
         this.emit('error', error);
+        const cause = error instanceof Error ? error : new Error(String(error));
+        void this.invalidateConnection(connection, cause).catch(() => {});
       });
 
       await this.withTimeout(

src/transport/stdio.ts

@@ -3,7 +3,7 @@
  */
 
 import { ChildProcess, spawn } from 'child_process';
-import { BaseTransport, TransportError } from './base.js';
+import { BaseTransport, TransportError, TransportState } from './base.js';
 import type { JsonRpcMessage } from '../types/jsonrpc.js';
 import type { TransportType } from '../types/service.js';
 import * as log from '../utils/logger.js';
@@ -246,16 +246,22 @@ export class StdioTransport extends BaseTransport {
 
   /**
    * Handle process exit
+   *
+   * Any exit not caused by our own close() means the backend process is gone and
+   * the connection is dead — even a clean code-0 exit or an external signal like
+   * SIGTERM/SIGKILL. Mark the transport ERROR so the connection pool stops reusing
+   * it. During an intentional close() the state is CLOSING/CLOSED, so skip then.
    */
   private handleProcessExit(code: number | null, signal: NodeJS.Signals | null): void {
     const exitInfo = signal ? `signal ${signal}` : `code ${code}`;
 
-    // Only treat non-zero exit codes as errors
-    // Signals like SIGTERM are normal termination
-    if (code !== null && code !== 0) {
+    if (this.state !== TransportState.CLOSING && this.state !== TransportState.CLOSED) {
       this.handleError(new TransportError(`Process exited with ${exitInfo}`, 'PROCESS_EXITED'));
     }
 
+    // Close the receive side and settle all waiting receivers
+    this.handleStreamEnd();
+
     // Reject all waiting receivers
     const error = new TransportError(`Process exited with ${exitInfo}`, 'PROCESS_EXITED');
     while (this.rejectQueue.length > 0) {

tests/integration/backend-recovery.test.ts

@@ -0,0 +1,198 @@
+/**
+ * Integration tests: backend death does not permanently break tool discovery.
+ *
+ * These tests drive the REAL ConnectionPool with a REAL StdioTransport that
+ * spawns a real backend process (the mock-stdio-mcp fixture). They cover the
+ * end-to-end recovery chain the unit tests mock away:
+ *
+ *   backend process dies → transport reaches ERROR → isConnected() flips false
+ *   → pool invalidates the connection → next acquire creates a fresh connection
+ *   → tools/list succeeds again
+ *
+ * This is the actual user-facing scenario from MERC-3: "long-running server
+ * mode loses tools after the backend connection drops".
+ */
+
+import { describe, it, expect, beforeEach, afterEach, vi } from 'vitest';
+import * as path from 'path';
+import { ConnectionPool } from '../../src/pool/connection-pool.js';
+import type { Connection } from '../../src/pool/connection.js';
+import type { JsonRpcMessage } from '../../src/types/jsonrpc.js';
+import type { ServiceDefinition, ConnectionPoolConfig } from '../../src/types/service.js';
+
+const FIXTURE_PATH = path.resolve(__dirname, 'fixtures/mock-stdio-mcp.cjs');
+
+const EXPECTED_TOOL_COUNT = 2;
+
+/**
+ * Send a tools/list request over a connection and return the tools array.
+ * Throws if the transport returns an error or the stream ends first — which is
+ * exactly the TRANSPORT_ERROR-equivalent failure the fix is meant to prevent.
+ */
+async function listTools(connection: Connection): Promise<unknown[]> {
+  const id = `tools-list-${Date.now()}-${Math.random().toString(36).slice(2)}`;
+  await connection.transport.send({
+    jsonrpc: '2.0',
+    id,
+    method: 'tools/list',
+    params: {},
+  } as JsonRpcMessage);
+
+  const iterator = connection.transport.receive();
+  try {
+    for (;;) {
+      const next = await iterator.next();
+      if (next.done || !next.value) {
+        throw new Error('transport stream ended before tools/list response');
+      }
+      const message = next.value as unknown as Record<string, unknown>;
+      if ('id' in message && String(message['id']) === String(id)) {
+        const error = message['error'] as { code?: number; message?: string } | undefined;
+        if (error) {
+          throw new Error(`tools/list returned error code ${error.code}: ${error.message}`);
+        }
+        const result = message['result'] as { tools?: unknown[] } | undefined;
+        return result?.tools ?? [];
+      }
+    }
+  } finally {
+    await iterator.return?.(undefined as unknown as JsonRpcMessage);
+  }
+}
+
+function waitForExit(pool: ConnectionPool, timeoutMs = 5000): Promise<void> {
+  return new Promise((resolve, reject) => {
+    const timer = setTimeout(() => {
+      pool.off('connectionFailed', onFailed);
+      reject(new Error('Timed out waiting for backend process exit / pool invalidation'));
+    }, timeoutMs);
+    const onFailed = () => {
+      clearTimeout(timer);
+      pool.off('connectionFailed', onFailed);
+      resolve();
+    };
+    pool.on('connectionFailed', onFailed);
+  });
+}
+
+describe('Backend death recovery (integration)', () => {
+  let pool: ConnectionPool;
+  let service: ServiceDefinition;
+  let poolConfig: ConnectionPoolConfig;
+
+  beforeEach(() => {
+    // Swallow pool-level error re-emits (transport errors surface here too).
+    vi.spyOn(console, 'error').mockImplementation(() => {});
+
+    service = {
+      name: 'mock-backend',
+      enabled: true,
+      tags: [],
+      transport: 'stdio',
+      command: process.execPath,
+      args: [FIXTURE_PATH],
+      env: { ONEMCP_FIXTURE_EXIT_AFTER_LIST: '1' },
+      connectionPool: {
+        maxConnections: 2,
+        idleTimeout: 60000,
+        connectionTimeout: 10000,
+      },
+    };
+
+    poolConfig = {
+      maxConnections: 2,
+      idleTimeout: 60000,
+      connectionTimeout: 10000,
+    };
+
+    pool = new ConnectionPool(service, poolConfig);
+    pool.on('error', () => {});
+  });
+
+  afterEach(async () => {
+    if (pool) {
+      await pool.closeAll().catch(() => {});
+    }
+    vi.restoreAllMocks();
+  });
+
+  it('recovers tools/list after the stdio backend exits with code 0', async () => {
+    // First acquire: pool spawns backend A, runs the MCP handshake, returns conn1.
+    const conn1 = await pool.acquire();
+    const tools1 = await listTools(conn1);
+    expect(tools1).toHaveLength(EXPECTED_TOOL_COUNT);
+
+    // The fixture exits with code 0 right after responding — wait for the pool
+    // to invalidate the now-dead connection.
+    await waitForExit(pool);
+
+    // The dead connection must no longer occupy a pool slot.
+    expect(pool.getStats().total).toBe(0);
+
+    // Second acquire must return a FRESH connection (a new backend process),
+    // not the dead one — and tools/list must succeed again, not TRANSPORT_ERROR.
+    const conn2 = await pool.acquire();
+    expect(conn2.id).not.toBe(conn1.id);
+    expect(pool.isConnectionHealthy(conn2)).toBe(true);
+
+    const tools2 = await listTools(conn2);
+    expect(tools2).toHaveLength(EXPECTED_TOOL_COUNT);
+  }, 30000);
+
+  it('recovers tools/list after the stdio backend is killed by a signal', async () => {
+    // Use a backend that stays alive so we control the death moment precisely.
+    const liveService: ServiceDefinition = {
+      ...service,
+      env: {}, // no EXIT_AFTER_LIST -> stays alive
+    };
+    const livePool = new ConnectionPool(liveService, poolConfig);
+    livePool.on('error', () => {});
+    try {
+      const conn1 = await livePool.acquire();
+      const tools1 = await listTools(conn1);
+      expect(tools1).toHaveLength(EXPECTED_TOOL_COUNT);
+      livePool.release(conn1);
+
+      // Kill the backend child process externally (simulates OOM / signal kill).
+      const proc = (conn1.transport as unknown as { process: { kill: (s: string) => void } })
+        .process;
+      const exitPromise = waitForExit(livePool);
+      proc.kill('SIGKILL');
+
+      await exitPromise;
+      expect(livePool.getStats().total).toBe(0);
+
+      // Next acquire creates a fresh connection; tools/list succeeds again.
+      const conn2 = await livePool.acquire();
+      expect(conn2.id).not.toBe(conn1.id);
+      const tools2 = await listTools(conn2);
+      expect(tools2).toHaveLength(EXPECTED_TOOL_COUNT);
+    } finally {
+      await livePool.closeAll().catch(() => {});
+    }
+  }, 30000);
+
+  it('does not reuse a dead idle connection when maxConnections would be exhausted', async () => {
+    // With maxConnections=1, a dead idle connection must be evicted (not held)
+    // so the next acquire can create a replacement rather than hanging.
+    const singleConfig: ConnectionPoolConfig = { ...poolConfig, maxConnections: 1 };
+    const singlePool = new ConnectionPool(service, singleConfig);
+    singlePool.on('error', () => {});
+    try {
+      const conn1 = await singlePool.acquire();
+      await listTools(conn1);
+      singlePool.release(conn1);
+
+      await waitForExit(singlePool);
+      expect(singlePool.getStats().total).toBe(0);
+
+      // Would hang / time out if the dead connection kept its slot.
+      const conn2 = await singlePool.acquire();
+      const tools2 = await listTools(conn2);
+      expect(tools2).toHaveLength(EXPECTED_TOOL_COUNT);
+      expect(conn2.id).not.toBe(conn1.id);
+    } finally {
+      await singlePool.closeAll().catch(() => {});
+    }
+  }, 30000);
+});