diff --git a/.changeset/keepalive-lifecycle-hardening.md b/.changeset/keepalive-lifecycle-hardening.md new file mode 100644 index 0000000000..c7cf2940fe --- /dev/null +++ b/.changeset/keepalive-lifecycle-hardening.md @@ -0,0 +1,5 @@ +--- +'@modelcontextprotocol/sdk': patch +--- + +Hardens the Streamable HTTP server transport's SSE lifecycle: deferred work cannot register streams after transport close, error cleanup preserves successor request mappings, invalid timer delays safely disable keep-alive, and SSE responses disable proxy buffering. diff --git a/src/server/webStandardStreamableHttp.ts b/src/server/webStandardStreamableHttp.ts index 932ad56600..f8946f8778 100644 --- a/src/server/webStandardStreamableHttp.ts +++ b/src/server/webStandardStreamableHttp.ts @@ -156,7 +156,8 @@ export interface WebStandardStreamableHTTPServerTransportOptions { * * Comment frames are ignored by SSE parsers and never surface as messages. * Defaults to 15000 (per the WHATWG SSE spec recommendation of roughly every - * 15 seconds). Set to 0 to disable keep-alive frames. + * 15 seconds). Set to 0 to disable keep-alive frames; values below 1, above + * 2147483647, or non-finite values also disable the timer. */ keepAliveMs?: number; } @@ -227,6 +228,7 @@ export class WebStandardStreamableHTTPServerTransport implements Transport { // when sessionId is not set (undefined), it means the transport is in stateless mode private sessionIdGenerator: (() => string) | undefined; private _started: boolean = false; + private _closed: boolean = false; private _hasHandledRequest: boolean = false; private _streamMapping: Map = new Map(); private _requestToStreamMapping: Map = new Map(); @@ -271,7 +273,11 @@ export class WebStandardStreamableHTTPServerTransport implements Transport { * clears itself if a write fails (stream already closed/cancelled). */ private startKeepAlive(streamId: string, controller: ReadableStreamDefaultController, encoder: TextEncoder): void { - if (this._keepAliveMs <= 0) { + // A deferred arm (e.g. after an event-store await that straddled + // close()) must not outlive the transport: close()'s timer sweep has + // already run. Invalid timer delays disable keep-alive rather than + // letting setInterval clamp them to ~1ms and flood every stream. + if (!Number.isFinite(this._keepAliveMs) || this._keepAliveMs < 1 || this._keepAliveMs > 2_147_483_647 || this._closed) { return; } this.stopKeepAlive(streamId); @@ -376,6 +382,10 @@ export class WebStandardStreamableHTTPServerTransport implements Transport { * Returns a Response object (Web Standard) */ async handleRequest(req: Request, options?: HandleRequestOptions): Promise { + if (this._closed) { + return this.createJsonErrorResponse(404, -32001, 'Session not found'); + } + // In stateless mode (no sessionIdGenerator), each request must use a fresh transport. // Reusing a stateless transport causes message ID collisions between clients. if (!this.sessionIdGenerator && this._hasHandledRequest) { @@ -488,7 +498,8 @@ export class WebStandardStreamableHTTPServerTransport implements Transport { const headers: Record = { 'Content-Type': 'text/event-stream', 'Cache-Control': 'no-cache, no-transform', - Connection: 'keep-alive' + Connection: 'keep-alive', + 'X-Accel-Buffering': 'no' }; // After initialization, always include the session ID if we have one @@ -547,7 +558,8 @@ export class WebStandardStreamableHTTPServerTransport implements Transport { const headers: Record = { 'Content-Type': 'text/event-stream', 'Cache-Control': 'no-cache, no-transform', - Connection: 'keep-alive' + Connection: 'keep-alive', + 'X-Accel-Buffering': 'no' }; if (this.sessionId !== undefined) { @@ -583,6 +595,21 @@ export class WebStandardStreamableHTTPServerTransport implements Transport { } }); + // The transport may have closed while the replay await was parked: + // its cleanup sweep ran before this stream was registered, so + // registering now would strand a mapping (and an open controller) + // on a dead transport, hang the client on a stream that never + // ends, and 409-block a later resume of this stream id. End the + // stream instead so the client observes termination. + if (this._closed) { + try { + streamController!.close(); + } catch { + // Controller might already be closed + } + return new Response(readable, { headers }); + } + this._streamMapping.set(replayedStreamId, { controller: streamController!, encoder, @@ -658,6 +685,12 @@ export class WebStandardStreamableHTTPServerTransport implements Transport { * Handles POST requests containing JSON-RPC messages */ private async handlePostRequest(req: Request, options?: HandleRequestOptions): Promise { + // Set once the SSE stream bookkeeping has been registered, so the + // catch below can reclaim it: an error after registration (a failed + // priming event write, a throwing message handler) returns an error + // response, leaving nothing that could ever cancel the discarded + // stream or retire the request mappings. + let reclaimSseBookkeeping: (() => void) | undefined; try { // Validate the Accept header const acceptHeader = req.headers.get('accept'); @@ -750,6 +783,12 @@ export class WebStandardStreamableHTTPServerTransport implements Transport { } } + // Request parsing and session initialization may await user/runtime + // work. Do not register or dispatch after close() has swept state. + if (this._closed) { + return this.createJsonErrorResponse(404, -32001, 'Session not found'); + } + // check if it contains requests const hasRequests = messages.some(isJSONRPCRequest); @@ -812,8 +851,9 @@ export class WebStandardStreamableHTTPServerTransport implements Transport { const headers: Record = { 'Content-Type': 'text/event-stream', - 'Cache-Control': 'no-cache', - Connection: 'keep-alive' + 'Cache-Control': 'no-cache, no-transform', + Connection: 'keep-alive', + 'X-Accel-Buffering': 'no' }; // After initialization, always include the session ID if we have one @@ -842,7 +882,14 @@ export class WebStandardStreamableHTTPServerTransport implements Transport { } } - this.startKeepAlive(streamId, streamController!, encoder); + reclaimSseBookkeeping = () => { + this._streamMapping.get(streamId)?.cleanup(); + for (const message of messages) { + if (isJSONRPCRequest(message) && this._requestToStreamMapping.get(message.id) === streamId) { + this._requestToStreamMapping.delete(message.id); + } + } + }; // Write priming event if event store is configured (after mapping is set up) await this.writePrimingEvent(streamController!, encoder, streamId, clientProtocolVersion); @@ -869,10 +916,19 @@ export class WebStandardStreamableHTTPServerTransport implements Transport { // The server SHOULD NOT close the SSE stream before sending all JSON-RPC responses // This will be handled by the send() method when responses are ready + // Arm keep-alive only after the fallible awaits above — an error + // path returning 400 discards the Response, so nothing could ever + // cancel the stream and clear an already-armed timer. Skip if the + // responses already completed and cleaned the stream up. + if (this._streamMapping.get(streamId)?.controller === streamController!) { + this.startKeepAlive(streamId, streamController!, encoder); + } + return new Response(readable, { status: 200, headers }); } catch (error) { // return JSON-RPC formatted error this.onerror?.(error as Error); + reclaimSseBookkeeping?.(); return this.createJsonErrorResponse(400, -32700, 'Parse error', { data: String(error) }); } } @@ -961,6 +1017,8 @@ export class WebStandardStreamableHTTPServerTransport implements Transport { } async close(): Promise { + this._closed = true; + // Close all SSE connections this._streamMapping.forEach(({ cleanup }) => { cleanup(); diff --git a/test/server/streamableHttp.test.ts b/test/server/streamableHttp.test.ts index 2046a71697..ac4a457248 100644 --- a/test/server/streamableHttp.test.ts +++ b/test/server/streamableHttp.test.ts @@ -270,6 +270,8 @@ describe.each(zodTestMatrix)('$zodVersionLabel', (entry: ZodMatrixEntry) => { expect(response.status).toBe(200); expect(response.headers.get('content-type')).toBe('text/event-stream'); + expect(response.headers.get('cache-control')).toBe('no-cache, no-transform'); + expect(response.headers.get('x-accel-buffering')).toBe('no'); expect(response.headers.get('mcp-session-id')).toBeDefined(); }); @@ -486,6 +488,8 @@ describe.each(zodTestMatrix)('$zodVersionLabel', (entry: ZodMatrixEntry) => { expect(sseResponse.status).toBe(200); expect(sseResponse.headers.get('content-type')).toBe('text/event-stream'); + expect(sseResponse.headers.get('cache-control')).toBe('no-cache, no-transform'); + expect(sseResponse.headers.get('x-accel-buffering')).toBe('no'); // Send a notification (server-initiated message) that should appear on SSE stream const notification: JSONRPCMessage = { @@ -1444,6 +1448,8 @@ describe.each(zodTestMatrix)('$zodVersionLabel', (entry: ZodMatrixEntry) => { }); expect(reconnectResponse.status).toBe(200); + expect(reconnectResponse.headers.get('cache-control')).toBe('no-cache, no-transform'); + expect(reconnectResponse.headers.get('x-accel-buffering')).toBe('no'); // Read the replayed notification const reconnectReader = reconnectResponse.body?.getReader(); @@ -3290,6 +3296,10 @@ describe('WebStandardStreamableHTTPServerTransport SSE keep-alive', () => { }); } + function withSession(sessionId: string, extra?: Record): Record { + return { 'mcp-session-id': sessionId, 'mcp-protocol-version': '2025-11-25', ...extra }; + } + async function createTransport(options?: { keepAliveMs?: number }): Promise<{ transport: WebStandardStreamableHTTPServerTransport; sessionId: string; @@ -3447,4 +3457,199 @@ describe('WebStandardStreamableHTTPServerTransport SSE keep-alive', () => { await transport.close(); }); + + it.each([0.5, Number.NaN, Number.POSITIVE_INFINITY, 2_147_483_648])( + 'should disable keep-alive for invalid keepAliveMs %s instead of arming a clamped interval', + async keepAliveMs => { + const { transport, sessionId } = await createTransport({ keepAliveMs }); + + const response = await transport.handleRequest(req('GET', { headers: withSession(sessionId) })); + expect(response.status).toBe(200); + + // No timer may be armed: setInterval with a NaN/out-of-range delay + // is clamped by Node to ~1ms and would flood the stream with + // keep-alive frames. + expect(vi.getTimerCount()).toBe(0); + const reader = response.body!.getReader(); + await vi.advanceTimersByTimeAsync(60000); + const raced = await Promise.race([reader.read(), Promise.resolve('pending')]); + expect(raced).toBe('pending'); + + await transport.close(); + } + ); + + it('should not arm keep-alive when the transport closes during an event-store replay await', async () => { + let releaseReplay: (() => void) | undefined; + const eventStore: EventStore = { + async storeEvent(): Promise { + return 'evt-1'; + }, + async replayEventsAfter(): Promise { + await new Promise(resolve => { + releaseReplay = resolve; + }); + return '_GET_stream'; + } + }; + const transport = new WebStandardStreamableHTTPServerTransport({ sessionIdGenerator: () => randomUUID(), eventStore }); + await new McpServer({ name: 'test-server', version: '1.0.0' }).connect(transport); + const initResponse = await transport.handleRequest(req('POST', { body: TEST_MESSAGES.initialize })); + const sessionId = initResponse.headers.get('mcp-session-id') as string; + + // Enter replayEvents and park on the replayEventsAfter await + const pendingGet = transport.handleRequest(req('GET', { headers: { ...withSession(sessionId), 'Last-Event-ID': 'evt-1' } })); + await vi.advanceTimersByTimeAsync(0); + expect(releaseReplay).toBeDefined(); + + // Close the transport mid-await, then let the replay continuation run + await transport.close(); + releaseReplay?.(); + const replayResponse = await pendingGet; + + // The deferred continuation must not have armed a timer close() can never sweep + expect(vi.getTimerCount()).toBe(0); + + // The continuation must not re-register the stream on the closed + // transport: the client observes stream end instead of hanging on a + // dead session, and a later resume isn't 409-blocked by a stale entry. + const { done } = await replayResponse.body!.getReader().read(); + expect(done).toBe(true); + const internals = transport as unknown as { _streamMapping: Map }; + expect(internals._streamMapping.size).toBe(0); + }); + + it('should not leak a keep-alive timer when the priming event write fails on a POST SSE stream', async () => { + // Healthy during initialization, then the store starts failing — the + // tool call's priming event write must reject inside handlePostRequest + let storeFails = false; + const eventStore: EventStore = { + async storeEvent(): Promise { + if (storeFails) { + throw new Error('event store unavailable'); + } + return `evt-${randomUUID()}`; + }, + async replayEventsAfter(): Promise { + return 'stream-1'; + } + }; + const transport = new WebStandardStreamableHTTPServerTransport({ sessionIdGenerator: () => randomUUID(), eventStore }); + const mcpServer = new McpServer({ name: 'test-server', version: '1.0.0' }); + mcpServer.registerTool('noop', { description: 'noop' }, async () => ({ content: [] })); + await mcpServer.connect(transport); + + const initResponse = await transport.handleRequest(req('POST', { body: TEST_MESSAGES.initialize })); + expect(initResponse.status).toBe(200); + const sessionId = initResponse.headers.get('mcp-session-id') as string; + // Let the init response finish sending (its send() stores an event and + // then cleans up the init stream's keep-alive) before failing the store + await vi.advanceTimersByTimeAsync(0); + expect(vi.getTimerCount()).toBe(0); + storeFails = true; + + await transport.handleRequest( + req('POST', { + body: { jsonrpc: '2.0', method: 'tools/call', params: { name: 'noop', arguments: {} }, id: 'call-1' }, + headers: withSession(sessionId) + }) + ); + + // The discarded stream must not carry a permanently-firing timer + expect(vi.getTimerCount()).toBe(0); + + // The stream bookkeeping registered before the failed priming write + // must be reclaimed too: repeated failures during an event-store + // outage must not accrete orphaned stream entries or request mappings. + const internals = transport as unknown as { + _streamMapping: Map; + _requestToStreamMapping: Map; + }; + expect(internals._streamMapping.size).toBe(0); + expect(internals._requestToStreamMapping.size).toBe(0); + + await transport.close(); + }); + + it('should not reclaim a successor mapping that reused the failed POST request id', async () => { + let failPriming = false; + let rejectPriming: (() => void) | undefined; + const eventStore: EventStore = { + async storeEvent(): Promise { + if (failPriming) { + return new Promise((_resolve, reject) => { + rejectPriming = () => reject(new Error('event store unavailable')); + }); + } + return `evt-${randomUUID()}`; + }, + async replayEventsAfter(): Promise { + return 'stream-1'; + } + }; + const transport = new WebStandardStreamableHTTPServerTransport({ sessionIdGenerator: () => randomUUID(), eventStore }); + const mcpServer = new McpServer({ name: 'test-server', version: '1.0.0' }); + mcpServer.registerTool('noop', { description: 'noop' }, async () => ({ content: [] })); + await mcpServer.connect(transport); + + const initResponse = await transport.handleRequest(req('POST', { body: TEST_MESSAGES.initialize })); + const sessionId = initResponse.headers.get('mcp-session-id') as string; + await vi.advanceTimersByTimeAsync(0); + failPriming = true; + + const pending = transport.handleRequest( + req('POST', { + body: { jsonrpc: '2.0', method: 'tools/call', params: { name: 'noop', arguments: {} }, id: 'same-id' }, + headers: withSession(sessionId) + }) + ); + await vi.advanceTimersByTimeAsync(0); + expect(rejectPriming).toBeDefined(); + + const internals = transport as unknown as { + _streamMapping: Map; + _requestToStreamMapping: Map; + }; + const failedStreamId = internals._requestToStreamMapping.get('same-id'); + expect(failedStreamId).toBeDefined(); + internals._requestToStreamMapping.set('same-id', 'successor-stream'); + + rejectPriming?.(); + await pending; + + expect(internals._streamMapping.has(failedStreamId!)).toBe(false); + expect(internals._requestToStreamMapping.get('same-id')).toBe('successor-stream'); + internals._requestToStreamMapping.delete('same-id'); + await transport.close(); + }); + + it('should not register a POST stream after close races session initialization', async () => { + let releaseInitialization: (() => void) | undefined; + const transport = new WebStandardStreamableHTTPServerTransport({ + sessionIdGenerator: () => randomUUID(), + onsessioninitialized: async () => { + await new Promise(resolve => { + releaseInitialization = resolve; + }); + } + }); + await new McpServer({ name: 'test-server', version: '1.0.0' }).connect(transport); + + const pendingInit = transport.handleRequest(req('POST', { body: TEST_MESSAGES.initialize })); + await vi.advanceTimersByTimeAsync(0); + expect(releaseInitialization).toBeDefined(); + + await transport.close(); + releaseInitialization?.(); + const response = await pendingInit; + + expect(response.status).toBe(404); + expect(vi.getTimerCount()).toBe(0); + const internals = transport as unknown as { + _streamMapping: Map; + _requestToStreamMapping: Map; + }; + expect(internals._streamMapping.size).toBe(0); + expect(internals._requestToStreamMapping.size).toBe(0); + }); });