fix: harden SSE keep-alive lifecycle (v1.x backport of the #2541 review fixes)

.changeset/keepalive-lifecycle-hardening.md

@@ -0,0 +1,5 @@
+---
+'@modelcontextprotocol/sdk': patch
+---
+
+Hardens the Streamable HTTP server transport's SSE keep-alive lifecycle: timers can no longer be armed after transport close or leak when a priming event write fails, invalid timer delays safely disable keep-alive, and SSE responses disable nginx-style proxy buffering.

src/server/webStandardStreamableHttp.ts

@@ -156,7 +156,8 @@
      *
      * 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 @@
     // 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<string, StreamMapping> = new Map();
     private _requestToStreamMapping: Map<RequestId, string> = new Map();
@@ -271,7 +273,11 @@
      * clears itself if a write fails (stream already closed/cancelled).
      */
     private startKeepAlive(streamId: string, controller: ReadableStreamDefaultController<Uint8Array>, 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);
@@ -488,7 +494,8 @@
         const headers: Record<string, string> = {
             '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 +554,8 @@
             const headers: Record<string, string> = {
                 '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 +591,21 @@
                 }
             });
 
+            // 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 +681,12 @@
      * Handles POST requests containing JSON-RPC messages
      */
     private async handlePostRequest(req: Request, options?: HandleRequestOptions): Promise<Response> {
+        // 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');
@@ -810,11 +839,12 @@
                 }
             });
 
             const headers: Record<string, string> = {
                 'Content-Type': 'text/event-stream',
                 'Cache-Control': 'no-cache',
-                Connection: 'keep-alive'
+                Connection: 'keep-alive',
+                'X-Accel-Buffering': 'no'
             };
 
             // After initialization, always include the session ID if we have one
             if (this.sessionId !== undefined) {
@@ -842,7 +872,14 @@
                 }
             }
 
-            this.startKeepAlive(streamId, streamController!, encoder);
+            reclaimSseBookkeeping = () => {
+                this._streamMapping.get(streamId)?.cleanup();
+                for (const message of messages) {
+                    if (isJSONRPCRequest(message)) {
+                        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 +906,19 @@
             // 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 +1007,8 @@
     }
 
     async close(): Promise<void> {
+        this._closed = true;
+
         // Close all SSE connections
         this._streamMapping.forEach(({ cleanup }) => {
             cleanup();

test/server/streamableHttp.test.ts

@@ -270,6 +270,7 @@ 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('x-accel-buffering')).toBe('no');
             expect(response.headers.get('mcp-session-id')).toBeDefined();
         });
 
@@ -486,6 +487,7 @@ 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('x-accel-buffering')).toBe('no');
 
             // Send a notification (server-initiated message) that should appear on SSE stream
             const notification: JSONRPCMessage = {
@@ -1444,6 +1446,7 @@ describe.each(zodTestMatrix)('$zodVersionLabel', (entry: ZodMatrixEntry) => {
             });
 
             expect(reconnectResponse.status).toBe(200);
+            expect(reconnectResponse.headers.get('x-accel-buffering')).toBe('no');
 
             // Read the replayed notification
             const reconnectReader = reconnectResponse.body?.getReader();
@@ -3290,6 +3293,10 @@ describe('WebStandardStreamableHTTPServerTransport SSE keep-alive', () => {
         });
     }
 
+    function withSession(sessionId: string, extra?: Record<string, string>): Record<string, string> {
+        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 +3454,117 @@ 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<EventId> {
+                return 'evt-1';
+            },
+            async replayEventsAfter(): Promise<StreamId> {
+                await new Promise<void>(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<string, unknown> };
+        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<EventId> {
+                if (storeFails) {
+                    throw new Error('event store unavailable');
+                }
+                return `evt-${randomUUID()}`;
+            },
+            async replayEventsAfter(): Promise<StreamId> {
+                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<string, unknown>;
+            _requestToStreamMapping: Map<unknown, string>;
+        };
+        expect(internals._streamMapping.size).toBe(0);
+        expect(internals._requestToStreamMapping.size).toBe(0);
+
+        await transport.close();
+    });
 });