From 0b9d5b09dc5401b1fd93cc95d59ee136fdb9eb92 Mon Sep 17 00:00:00 2001 From: Aaron Bockelie Date: Wed, 24 Jun 2026 01:45:25 -0500 Subject: [PATCH 1/2] fix(mcp): stop reaping SSE notify stream; drop dead transport handlers MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two transport/session-lifecycle hygiene fixes surfaced while tracing the #221/#238 session cluster. SSE notify-stream churn: `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 set in start() reaped it every ~2 min, surfacing as the "stream terminated" reconnect churn and noisy logs in bridge reports (#221). handleMCPRequest now exempts only the GET (SSE) socket from the idle timeout via req.socket.setTimeout(0); POST request sockets keep the server-wide default. Dead code: removed attachTransportHandlers, which wired transport.on('close'|'error'). SDK 1.29's StreamableHTTPServerTransport is not an EventEmitter and has no `.on`, so it never fired โ€” dead and misleading. Transport cleanup is already covered by the SessionManager `session-evicted` handler (close + map delete; every transport maps to a manager-tracked session that is eventually idle-evicted, so none leaks) and the explicit DELETE /mcp handler. Replaced the helper with a comment documenting those two paths. Adds tests/sse-socket-timeout.test.ts (GET exempts its socket, POST does not, missing-socket tolerated). Refs #221. Behavioral SSE change; verify on a live Obsidian instance before release. Claude-Session: https://claude.ai/code/session_011yowKCMFCBp61DRBu5xQm4 --- src/mcp-server.ts | 48 ++++++++--------- tests/sse-socket-timeout.test.ts | 88 ++++++++++++++++++++++++++++++++ 2 files changed, 109 insertions(+), 27 deletions(-) create mode 100644 tests/sse-socket-timeout.test.ts diff --git a/src/mcp-server.ts b/src/mcp-server.ts index 40cedf1..de5df37 100644 --- a/src/mcp-server.ts +++ b/src/mcp-server.ts @@ -438,6 +438,16 @@ 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). Exempt only this socket from the idle + // timeout; POST request sockets keep the server-wide default. + 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 +466,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 +502,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 +534,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..9fd1cec --- /dev/null +++ b/tests/sse-socket-timeout.test.ts @@ -0,0 +1,88 @@ +/** + * 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); + }); + + 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 (no throw)', async () => { + const req = { method: 'GET', body: undefined, headers: {} as Record }; + const res = makeRes(); + await expect((server as any).handleMCPRequest(req, res)).resolves.toBeUndefined(); + }); +}); From 926b8fb363b24df795d19bd65168fa3a5d098a91 Mon Sep 17 00:00:00 2001 From: Aaron Bockelie Date: Wed, 24 Jun 2026 02:00:33 -0500 Subject: [PATCH 2/2] =?UTF-8?q?refactor(mcp):=20address=20review=20of=20#2?= =?UTF-8?q?44=20=E2=80=94=20document=20SSE=20backstop,=20test=20teardown?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Code review refinements (no behavior change): - Make the SSE socket-exemption trade-off explicit: with setTimeout(0) there is no socket-layer idle reap, so a truly-dead SSE socket is bounded by the SessionManager's 1h idle eviction (which calls transport.close()) and the 32-session cap, not leaked unbounded. Note the finite-timeout alternative is deferred to live validation. - Clarify the comment: the exemption matches any GET, but non-SSE GETs are short-lived so it's a harmless no-op for them. - Add afterEach teardown disposing the SessionManager + connection-pool intervals (server.stop() short-circuits when never started), removing the test's leaked-timer warning. - Strengthen the missing-socket test to assert the request still routes to the spec ยง2 400, not just that it doesn't throw. Claude-Session: https://claude.ai/code/session_011yowKCMFCBp61DRBu5xQm4 --- src/mcp-server.ts | 15 +++++++++++++-- tests/sse-socket-timeout.test.ts | 15 +++++++++++++-- 2 files changed, 26 insertions(+), 4 deletions(-) diff --git a/src/mcp-server.ts b/src/mcp-server.ts index de5df37..0ba69e3 100644 --- a/src/mcp-server.ts +++ b/src/mcp-server.ts @@ -443,8 +443,19 @@ export class MCPHttpServer { // 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). Exempt only this socket from the idle - // timeout; POST request sockets keep the server-wide default. + // 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); } diff --git a/tests/sse-socket-timeout.test.ts b/tests/sse-socket-timeout.test.ts index 9fd1cec..0882b43 100644 --- a/tests/sse-socket-timeout.test.ts +++ b/tests/sse-socket-timeout.test.ts @@ -68,6 +68,14 @@ describe('SSE notification-stream socket timeout exemption (#221)', () => { 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()); @@ -80,9 +88,12 @@ describe('SSE notification-stream socket timeout exemption (#221)', () => { expect(socket.setTimeout).not.toHaveBeenCalled(); }); - test('exemption tolerates a missing socket (no throw)', async () => { + 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 expect((server as any).handleMCPRequest(req, res)).resolves.toBeUndefined(); + 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); }); });