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: keep-alive timers can no longer be armed after the transport closes or leak when a priming event write fails, and a non-finite `keepAliveMs` disables keep-alive instead of arming a clamped ~1ms interval.

src/server/webStandardStreamableHttp.ts

@@ -227,6 +227,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();
@@ -268,13 +269,18 @@
      * Replaces any timer already armed for the same stream id (a resumed stream
      * re-registered under the same id supersedes its predecessor's timer). The
      * timer is cleared via stopKeepAlive when the stream is cleaned up, and
      * 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. The `> 0` polarity disables keep-alive for non-finite
+        // values (NaN fails every comparison) instead of arming a
+        // Node-clamped ~1ms interval.
+        if (!(this._keepAliveMs > 0) || this._closed) {
             return;
         }
         this.stopKeepAlive(streamId);
         const timer = setInterval(() => {
             try {
                 controller.enqueue(encoder.encode(': keepalive\n\n'));
@@ -842,8 +848,6 @@
                 }
             }
 
-            this.startKeepAlive(streamId, streamController!, encoder);
-
             // Write priming event if event store is configured (after mapping is set up)
             await this.writePrimingEvent(streamController!, encoder, streamId, clientProtocolVersion);
 
@@ -869,6 +873,14 @@
             // 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
@@ -961,6 +973,8 @@
     }
 
     async close(): Promise<void> {
+        this._closed = true;
+
         // Close all SSE connections
         this._streamMapping.forEach(({ cleanup }) => {
             cleanup();

test/server/streamableHttp.test.ts

@@ -3290,6 +3290,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 +3451,96 @@ describe('WebStandardStreamableHTTPServerTransport SSE keep-alive', () => {
 
         await transport.close();
     });
+
+    it('should disable keep-alive for a non-finite keepAliveMs instead of arming a clamped interval', async () => {
+        const { transport, sessionId } = await createTransport({ keepAliveMs: Number.NaN });
+
+        const response = await transport.handleRequest(req('GET', { headers: withSession(sessionId) }));
+        expect(response.status).toBe(200);
+
+        // No timer may be armed: setInterval(fn, NaN) would be clamped by
+        // Node to ~1ms and 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?.();
+        await pendingGet;
+
+        // The deferred continuation must not have armed a timer close() can never sweep
+        expect(vi.getTimerCount()).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;
+
+        const response = await transport.handleRequest(
+            req('POST', {
+                body: { jsonrpc: '2.0', method: 'tools/call', params: { name: 'noop', arguments: {} }, id: 'call-1' },
+                headers: withSession(sessionId)
+            })
+        );
+        expect(response.status).toBe(400);
+
+        // The discarded stream must not carry a permanently-firing timer
+        expect(vi.getTimerCount()).toBe(0);
+
+        await transport.close();
+    });
 });