diff --git a/src/mcp-server.ts b/src/mcp-server.ts index 40cedf1..0ba69e3 100644 --- a/src/mcp-server.ts +++ b/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 diff --git a/tests/sse-socket-timeout.test.ts b/tests/sse-socket-timeout.test.ts new file mode 100644 index 0000000..0882b43 --- /dev/null +++ b/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; + 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 }; + 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); + }); +});