Merge pull request #244 from aaronsb/fix/sse-noise-dead-transport-handlers

fix(mcp): stop reaping SSE notify stream; drop dead transport handlers

src/mcp-server.ts

@@ -438,6 +438,27 @@ export class MCPHttpServer {
       // Get or create session ID
       const sessionId = req.headers['mcp-session-id'] as string | undefined;
       Debug.log(`📨 MCP Request: ${request?.method ?? 'unknown'}${sessionId ? ` [Session: ${sessionId}]` : ''}`, request?.params);
+
+      // `GET /mcp` opens the standalone SSE notification stream — long-lived and
+      // idle by design (this server pushes no server-initiated notifications).
+      // The global 120s socket timeout set in start() would otherwise reap it
+      // every ~2 min, surfacing as a "stream terminated" reconnect churn in
+      // bridges and noisy logs (#221). Disable the idle timeout on this GET
+      // socket; POST request sockets keep the server-wide default. (A non-SSE
+      // GET is short-lived — its response is sent immediately — so the
+      // exemption is a harmless no-op for those.)
+      //
+      // Trade-off: with no socket-layer idle reap, a truly-dead SSE socket
+      // (client vanished without a FIN) is no longer dropped at ~2 min. The
+      // backstop is the SessionManager's 1h idle eviction, which calls
+      // transport.close() and thereby closes the socket — so abandoned streams
+      // are bounded by that timeout and the 32-session cap, not leaked
+      // unbounded. A long finite timeout was weighed against 0; the choice is
+      // deferred to the live-instance validation, since a finite value would
+      // reintroduce some (less frequent) reconnect churn.
+      if (req.method === 'GET') {
+        req.socket?.setTimeout(0);
+      }
       // Quick path: lightweight ping to keep session alive
       if (request?.method === 'session/ping' || request?.method === 'status/ping') {
         if (sessionId && this.sessionManager) {
@@ -456,31 +477,17 @@ export class MCPHttpServer {
       }
           let mcpServer: MCPServer;
 
-      // Helper: register transport with lifecycle hooks
-      const attachTransportHandlers = (sessId: string, tr: StreamableHTTPServerTransport) => {
-        try {
-          // Transport may optionally support EventEmitter API
-          const emitter = tr as unknown as { on?: (event: string, handler: (...args: unknown[]) => void) => void };
-          if (typeof emitter.on === 'function') {
-            emitter.on('close', () => {
-              if (this.transports.has(sessId)) {
-                this.transports.delete(sessId);
-                this.connectionCount = Math.max(0, this.connectionCount - 1);
-                Debug.log(`🔌 Transport closed for session ${sessId}. Connections: ${this.connectionCount}`);
-              }
-            });
-            emitter.on('error', (e: unknown) => {
-              Debug.error(`Transport error for session ${sessId}:`, e);
-              if (this.transports.has(sessId)) {
-                this.transports.delete(sessId);
-                this.connectionCount = Math.max(0, this.connectionCount - 1);
-              }
-            });
-          }
-        } catch {
-          // Transport may not support event emitters, which is fine
-        }
-      };
+      // Transport cleanup is handled by two existing paths, so there is no
+      // per-transport close hook here:
+      //   • idle eviction — the SessionManager emits `session-evicted`, whose
+      //     handler (see constructor) calls `transport.close()` and drops the
+      //     map entry; every transport maps to a manager-tracked session that
+      //     is eventually idle-evicted, so none leaks permanently.
+      //   • explicit teardown — the DELETE /mcp handler closes + removes it.
+      // A prior `transport.on('close'|'error')` helper here was dead code:
+      // SDK 1.29's StreamableHTTPServerTransport is not an EventEmitter and has
+      // no `.on`, so it never fired. Its only correct hook would be the
+      // `onclose` callback, which would merely duplicate the two paths above.
 
       // Determine which server to use from the pool
       if (sessionId && this.transports.has(sessionId)) {
@@ -506,7 +513,6 @@ export class MCPHttpServer {
             });
             await mcpServer.connect(transport);
             this.transports.set(effectiveSessionId, transport);
-            attachTransportHandlers(effectiveSessionId, transport);
             this.connectionCount++;
             Debug.log(`♻️ Recreated transport for session ${sessionId} (requests: ${session.requestCount})`);
           } else {
@@ -539,7 +545,6 @@ export class MCPHttpServer {
           
           // Store the transport for future requests
           this.transports.set(effectiveSessionId, transport);
-          attachTransportHandlers(effectiveSessionId, transport);
           this.connectionCount++;
           
           // Register session with manager if enabled

tests/sse-socket-timeout.test.ts

@@ -0,0 +1,99 @@
+/**
+ * Regression harness for the SSE notification-stream socket timeout (#221).
+ *
+ * `GET /mcp` opens the standalone SSE notification stream, which is long-lived
+ * and idle by design (this server pushes no server-initiated notifications).
+ * The server-wide 120s idle socket timeout (configured in start()) would reap
+ * that stream every ~2 min, surfacing as "stream terminated" reconnect churn in
+ * bridges and noisy logs. handleMCPRequest now exempts only the GET (SSE)
+ * socket from the idle timeout; POST request sockets keep the default.
+ *
+ * These assertions check the per-socket exemption directly — they return before
+ * any SDK transport work, so they're deterministic without driving the SDK
+ * transport in jsdom.
+ */
+import { MCPHttpServer } from '../src/mcp-server';
+import { App } from 'obsidian';
+
+jest.mock('fs', () => ({
+  existsSync: jest.fn(() => false),
+  mkdirSync: jest.fn(),
+  readFileSync: jest.fn(),
+  writeFileSync: jest.fn()
+}));
+
+interface MockRes {
+  statusCode: number;
+  headers: Record<string, string>;
+  body: unknown;
+  headersSent: boolean;
+  setHeader: (k: string, v: string) => void;
+  status: (c: number) => MockRes;
+  json: (b: unknown) => void;
+}
+
+function makeRes(): MockRes {
+  const res: MockRes = {
+    statusCode: 0,
+    headers: {},
+    body: undefined,
+    headersSent: false,
+    setHeader(k, v) { this.headers[k.toLowerCase()] = v; },
+    status(c) { this.statusCode = c; return this; },
+    json(b) { this.body = b; this.headersSent = true; }
+  };
+  return res;
+}
+
+function makeReq(method: string, body: unknown, sessionId?: string) {
+  const socket = { setTimeout: jest.fn() };
+  const req = {
+    method,
+    body,
+    headers: sessionId ? { 'mcp-session-id': sessionId } : {},
+    socket,
+  };
+  return { req, socket };
+}
+
+describe('SSE notification-stream socket timeout exemption (#221)', () => {
+  let server: MCPHttpServer;
+
+  beforeEach(() => {
+    const mockApp = new App();
+    mockApp.vault = {
+      ...mockApp.vault,
+      adapter: { basePath: '/mock/vault/path' }
+    } as any;
+    server = new MCPHttpServer(mockApp, 3001);
+  });
+
+  afterEach(async () => {
+    // The constructor arms the SessionManager + connection-pool intervals;
+    // server.stop() short-circuits when never started, so dispose them directly
+    // to avoid leaking timers across tests.
+    (server as any).sessionManager?.stop?.();
+    await (server as any).connectionPool?.shutdown?.();
+  });
+
+  test('GET /mcp (SSE stream) disables its socket idle timeout', async () => {
+    const { req, socket } = makeReq('GET', undefined, 'some-session');
+    await (server as any).handleMCPRequest(req, makeRes());
+    expect(socket.setTimeout).toHaveBeenCalledWith(0);
+  });
+
+  test('POST /mcp leaves the socket on the server-wide default timeout', async () => {
+    const { req, socket } = makeReq('POST', { jsonrpc: '2.0', id: 1, method: 'tools/call', params: {} }, 'some-session');
+    await (server as any).handleMCPRequest(req, makeRes());
+    expect(socket.setTimeout).not.toHaveBeenCalled();
+  });
+
+  test('exemption tolerates a missing socket and still routes the request', async () => {
+    const req = { method: 'GET', body: undefined, headers: {} as Record<string, string> };
+    const res = makeRes();
+    await (server as any).handleMCPRequest(req, res);
+    // No socket to exempt → optional chaining no-ops; the GET (no session, not
+    // an initialize) still routes to the spec §2 "session required" 400.
+    expect(res.statusCode).toBe(400);
+  });
+});