From 14b0b5fb2c97b72bd7351a406eb9e63bfaac67c8 Mon Sep 17 00:00:00 2001 From: Felix Weinberger Date: Fri, 24 Jul 2026 10:20:02 +0000 Subject: [PATCH 1/6] Fix SSE keep-alive timer lifecycle in Streamable HTTP server transport Keep-alive timers were tracked in a transport-level map keyed by stream id. The standalone GET stream and its resumed successors share one id, so a stale connection's cancel callback could stop the resumed stream's keep-alive and delete its stream mapping, and the interval's error handler could clear the wrong timer. Timers are now owned by the stream they were armed for: startKeepAlive returns the handle, the stream's own cancel/cleanup clears it, and the interval clears itself on write failure. Cancel callbacks only delete the stream mapping when it still points at their own stream. A resume closes the superseded stream cleanly, and a resume that completes after the transport closed (or after its stream was cancelled) returns 404 instead of registering a stream nothing can clean up. The POST path now arms keep-alive after its fallible awaits and releases the stream and request correlations if they fail. Non-finite keepAliveMs values disable keep-alive, and values above 2^31-1 are clamped instead of firing every millisecond. --- .changeset/sse-keepalive-timer-lifecycle.md | 7 + src/server/webStandardStreamableHttp.ts | 201 +++++++++++------ test/server/streamableHttp.test.ts | 231 +++++++++++++++++++- 3 files changed, 367 insertions(+), 72 deletions(-) create mode 100644 .changeset/sse-keepalive-timer-lifecycle.md diff --git a/.changeset/sse-keepalive-timer-lifecycle.md b/.changeset/sse-keepalive-timer-lifecycle.md new file mode 100644 index 0000000000..a0523949fd --- /dev/null +++ b/.changeset/sse-keepalive-timer-lifecycle.md @@ -0,0 +1,7 @@ +--- +'@modelcontextprotocol/sdk': patch +--- + +Fix SSE keep-alive timer lifecycle in the Streamable HTTP server transport. Keep-alive timers are now owned by the stream they were armed for, so a stale disconnect arriving after a reconnect no longer stops the live stream's keep-alive, a resume closes the superseded stream +cleanly, and a failed priming-event write no longer leaves a timer running or a stale request correlation behind. A resume that completes after the transport closed now gets a 404 instead of a silent stream nothing will ever write to. Non-finite `keepAliveMs` values now disable +keep-alive, and values above 2^31-1 are clamped instead of firing every millisecond. diff --git a/src/server/webStandardStreamableHttp.ts b/src/server/webStandardStreamableHttp.ts index 932ad56600..531a440c67 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. Non-finite values also + * disable; intervals above 2^31-1 ms are clamped to that maximum. */ keepAliveMs?: number; } @@ -242,7 +243,7 @@ export class WebStandardStreamableHTTPServerTransport implements Transport { private _enableDnsRebindingProtection: boolean; private _retryInterval?: number; private _keepAliveMs: number; - private _keepAliveTimers: Map> = new Map(); + private _closed = false; sessionId?: string; onclose?: () => void; @@ -265,37 +266,33 @@ export class WebStandardStreamableHTTPServerTransport implements Transport { /** * Arms a keep-alive interval for an SSE stream that periodically writes an SSE * comment frame so intermediaries and idle timeouts don't kill the connection. - * 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). + * The returned timer is owned by the stream it was armed for: the stream's + * cancel/cleanup callbacks must clear it. The interval clears itself if a + * write fails (stream already closed/cancelled). */ - private startKeepAlive(streamId: string, controller: ReadableStreamDefaultController, encoder: TextEncoder): void { - if (this._keepAliveMs <= 0) { - return; + private startKeepAlive( + controller: ReadableStreamDefaultController, + encoder: TextEncoder + ): ReturnType | undefined { + // Non-finite values disable keep-alive rather than arming a broken timer, + // and delays above 2^31-1 are clamped (setInterval treats them as 1ms). + // After close() nothing may arm: the stream cleanups have already run. + if (!Number.isFinite(this._keepAliveMs) || this._keepAliveMs <= 0 || this._closed) { + return undefined; } - this.stopKeepAlive(streamId); - const timer = setInterval(() => { - try { - controller.enqueue(encoder.encode(': keepalive\n\n')); - } catch { - this.stopKeepAlive(streamId); - } - }, this._keepAliveMs); + const timer = setInterval( + () => { + try { + controller.enqueue(encoder.encode(': keepalive\n\n')); + } catch { + clearInterval(timer); + } + }, + Math.min(this._keepAliveMs, 2 ** 31 - 1) + ); // Don't let the keep-alive timer hold the process open (Node.js only) (timer as { unref?: () => void }).unref?.(); - this._keepAliveTimers.set(streamId, timer); - } - - /** - * Clears the keep-alive interval for a stream, if one is armed. - */ - private stopKeepAlive(streamId: string): void { - const timer = this._keepAliveTimers.get(streamId); - if (timer !== undefined) { - clearInterval(timer); - this._keepAliveTimers.delete(streamId); - } + return timer; } /** @@ -472,6 +469,7 @@ export class WebStandardStreamableHTTPServerTransport implements Transport { const encoder = new TextEncoder(); let streamController: ReadableStreamDefaultController; + let keepAliveTimer: ReturnType | undefined = undefined; // Create a ReadableStream with a controller we can use to push SSE events const readable = new ReadableStream({ @@ -479,9 +477,15 @@ export class WebStandardStreamableHTTPServerTransport implements Transport { streamController = controller; }, cancel: () => { - // Stream was cancelled by client - this.stopKeepAlive(this._standaloneSseStreamId); - this._streamMapping.delete(this._standaloneSseStreamId); + // Stream was cancelled by client. Only tear down the mapping if it + // still points at this stream — a stale cancel arriving after a + // reconnect re-registered the id must not tear down the successor. + if (keepAliveTimer !== undefined) { + clearInterval(keepAliveTimer); + } + if (this._streamMapping.get(this._standaloneSseStreamId)?.controller === streamController) { + this._streamMapping.delete(this._standaloneSseStreamId); + } } }); @@ -501,7 +505,9 @@ export class WebStandardStreamableHTTPServerTransport implements Transport { controller: streamController!, encoder, cleanup: () => { - this.stopKeepAlive(this._standaloneSseStreamId); + if (keepAliveTimer !== undefined) { + clearInterval(keepAliveTimer); + } this._streamMapping.delete(this._standaloneSseStreamId); try { streamController!.close(); @@ -511,7 +517,7 @@ export class WebStandardStreamableHTTPServerTransport implements Transport { } }); - this.startKeepAlive(this._standaloneSseStreamId, streamController!, encoder); + keepAliveTimer = this.startKeepAlive(streamController!, encoder); return new Response(readable, { headers }); } @@ -557,19 +563,30 @@ export class WebStandardStreamableHTTPServerTransport implements Transport { // Create a ReadableStream with controller for SSE const encoder = new TextEncoder(); let streamController: ReadableStreamDefaultController; + let keepAliveTimer: ReturnType | undefined = undefined; + let replayedStreamId: string | undefined = undefined; + let cancelled = false; const readable = new ReadableStream({ start: controller => { streamController = controller; }, cancel: () => { - // Stream was cancelled by client - // Cleanup will be handled by the mapping + // Stream was cancelled by client. Only tear down the mapping if + // it still points at this stream — a stale cancel must not tear + // down a successor registered by a later resume. + cancelled = true; + if (keepAliveTimer !== undefined) { + clearInterval(keepAliveTimer); + } + if (replayedStreamId !== undefined && this._streamMapping.get(replayedStreamId)?.controller === streamController) { + this._streamMapping.delete(replayedStreamId); + } } }); // Replay events - returns the streamId for backwards compatibility - const replayedStreamId = await this._eventStore.replayEventsAfter(lastEventId, { + replayedStreamId = await this._eventStore.replayEventsAfter(lastEventId, { send: async (eventId: string, message: JSONRPCMessage) => { const success = this.writeSSEEvent(streamController!, encoder, message, eventId); if (!success) { @@ -583,12 +600,31 @@ export class WebStandardStreamableHTTPServerTransport implements Transport { } }); + // close() may have run, or the stream may have been cancelled, while + // the event store was replaying: don't hand out or register a stream + // that nothing can ever clean up again. + if (this._closed || cancelled) { + try { + streamController!.close(); + } catch { + // Controller might already be closed + } + return this.createJsonErrorResponse(404, -32001, 'Session not found'); + } + + // A reconnect re-registers the same stream id: close the predecessor + // stream cleanly (clearing its keep-alive timer) instead of leaving it + // registered nowhere with its timer still running. + this._streamMapping.get(replayedStreamId)?.cleanup(); + this._streamMapping.set(replayedStreamId, { controller: streamController!, encoder, cleanup: () => { - this.stopKeepAlive(replayedStreamId); - this._streamMapping.delete(replayedStreamId); + if (keepAliveTimer !== undefined) { + clearInterval(keepAliveTimer); + } + this._streamMapping.delete(replayedStreamId!); try { streamController!.close(); } catch { @@ -597,7 +633,7 @@ export class WebStandardStreamableHTTPServerTransport implements Transport { } }); - this.startKeepAlive(replayedStreamId, streamController!, encoder); + keepAliveTimer = this.startKeepAlive(streamController!, encoder); return new Response(readable, { headers }); } catch (error) { @@ -798,15 +834,22 @@ export class WebStandardStreamableHTTPServerTransport implements Transport { // SSE streaming mode - use ReadableStream with controller for more reliable data pushing const encoder = new TextEncoder(); let streamController: ReadableStreamDefaultController; + let keepAliveTimer: ReturnType | undefined = undefined; const readable = new ReadableStream({ start: controller => { streamController = controller; }, cancel: () => { - // Stream was cancelled by client - this.stopKeepAlive(streamId); - this._streamMapping.delete(streamId); + // Stream was cancelled by client. Only tear down the mapping if + // it still points at this stream — a stale cancel arriving after + // a resume re-registered the id must not tear down the successor. + if (keepAliveTimer !== undefined) { + clearInterval(keepAliveTimer); + } + if (this._streamMapping.get(streamId)?.controller === streamController) { + this._streamMapping.delete(streamId); + } } }); @@ -829,7 +872,9 @@ export class WebStandardStreamableHTTPServerTransport implements Transport { controller: streamController!, encoder, cleanup: () => { - this.stopKeepAlive(streamId); + if (keepAliveTimer !== undefined) { + clearInterval(keepAliveTimer); + } this._streamMapping.delete(streamId); try { streamController!.close(); @@ -842,33 +887,51 @@ export class WebStandardStreamableHTTPServerTransport implements Transport { } } - this.startKeepAlive(streamId, streamController!, encoder); + try { + // Write priming event if event store is configured (after mapping is set up) + await this.writePrimingEvent(streamController!, encoder, streamId, clientProtocolVersion); - // Write priming event if event store is configured (after mapping is set up) - await this.writePrimingEvent(streamController!, encoder, streamId, clientProtocolVersion); + // handle each message + for (const message of messages) { + // Build closeSSEStream callback for requests when eventStore is configured + // AND client supports resumability (protocol version >= 2025-11-25). + // Old clients can't resume if the stream is closed early because they + // didn't receive a priming event with an event ID. + let closeSSEStream: (() => void) | undefined; + let closeStandaloneSSEStream: (() => void) | undefined; + if (isJSONRPCRequest(message) && this._eventStore && clientProtocolVersion >= '2025-11-25') { + closeSSEStream = () => { + this.closeSSEStream(message.id); + }; + closeStandaloneSSEStream = () => { + this.closeStandaloneSSEStream(); + }; + } - // handle each message - for (const message of messages) { - // Build closeSSEStream callback for requests when eventStore is configured - // AND client supports resumability (protocol version >= 2025-11-25). - // Old clients can't resume if the stream is closed early because they - // didn't receive a priming event with an event ID. - let closeSSEStream: (() => void) | undefined; - let closeStandaloneSSEStream: (() => void) | undefined; - if (isJSONRPCRequest(message) && this._eventStore && clientProtocolVersion >= '2025-11-25') { - closeSSEStream = () => { - this.closeSSEStream(message.id); - }; - closeStandaloneSSEStream = () => { - this.closeStandaloneSSEStream(); - }; + this.onmessage?.(message, { authInfo: options?.authInfo, requestInfo, closeSSEStream, closeStandaloneSSEStream }); } - - this.onmessage?.(message, { authInfo: options?.authInfo, requestInfo, closeSSEStream, closeStandaloneSSEStream }); + } catch (error) { + // Failing before the Response is handed out means nothing can ever + // cancel this stream: release it and its request correlations here. + this._streamMapping.get(streamId)?.cleanup(); + for (const message of messages) { + if (isJSONRPCRequest(message)) { + this._requestToStreamMapping.delete(message.id); + } + } + throw error; } // 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 + // returns 400 and 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!) { + keepAliveTimer = this.startKeepAlive(streamController!, encoder); + } + return new Response(readable, { status: 200, headers }); } catch (error) { // return JSON-RPC formatted error @@ -961,16 +1024,16 @@ export class WebStandardStreamableHTTPServerTransport implements Transport { } async close(): Promise { - // Close all SSE connections + // Stop any deferred keep-alive arm (e.g. a replay await resolving after + // close) from creating a timer nothing can clear. + this._closed = true; + + // Close all SSE connections (each cleanup clears its own keep-alive timer) this._streamMapping.forEach(({ cleanup }) => { cleanup(); }); this._streamMapping.clear(); - // Clear any keep-alive timers not already cleared by stream cleanup - this._keepAliveTimers.forEach(timer => clearInterval(timer)); - this._keepAliveTimers.clear(); - // Clear any pending responses this._requestResponseMap.clear(); this.onclose?.(); diff --git a/test/server/streamableHttp.test.ts b/test/server/streamableHttp.test.ts index 2046a71697..83b6e8c30f 100644 --- a/test/server/streamableHttp.test.ts +++ b/test/server/streamableHttp.test.ts @@ -3412,9 +3412,9 @@ describe('WebStandardStreamableHTTPServerTransport SSE keep-alive', () => { it('should supersede the previous keep-alive timer when a replayed stream re-registers under the same stream id', async () => { // Event store WITHOUT the optional getStreamIdForEventId — the replay // path then skips its 409 conflict check, so a reconnect re-registers - // the same stream id. The predecessor's timer must be replaced, not - // orphaned (an orphaned timer's failing write would clear the live - // stream's keep-alive via stopKeepAlive on the shared stream id). + // the same stream id. The predecessor stream must be closed and its + // timer cleared, not left orphaned; each timer is owned by its stream, + // so a stale timer's failing write only ever clears itself. const eventStore: EventStore = { async storeEvent(): Promise { return 'evt-1'; @@ -3448,3 +3448,228 @@ describe('WebStandardStreamableHTTPServerTransport SSE keep-alive', () => { await transport.close(); }); }); + +describe('WebStandardStreamableHTTPServerTransport SSE keep-alive lifecycle', () => { + /** Shorthand to build a Web Standard Request for direct transport testing. */ + function req(method: string, opts?: { body?: unknown; headers?: Record }): Request { + const headers: Record = { ...opts?.headers }; + if (method === 'POST') { + headers['Accept'] ??= 'application/json, text/event-stream'; + headers['Content-Type'] ??= 'application/json'; + } else if (method === 'GET') { + headers['Accept'] ??= 'text/event-stream'; + } + return new Request('http://localhost/mcp', { + method, + headers, + body: opts?.body !== undefined ? JSON.stringify(opts.body) : undefined + }); + } + + /** + * Minimal event store WITHOUT the optional getStreamIdForEventId, so the + * replay path re-registers a stream id without a 409 conflict check — + * the reconnect shape these tests exercise. + */ + function createSimpleEventStore(): EventStore { + const events: { id: EventId; streamId: StreamId; message: JSONRPCMessage }[] = []; + let counter = 0; + return { + async storeEvent(streamId: StreamId, message: JSONRPCMessage): Promise { + const id = `${streamId}#${counter++}`; + events.push({ id, streamId, message }); + return id; + }, + async replayEventsAfter(lastEventId: EventId, { send }): Promise { + const index = events.findIndex(e => e.id === lastEventId); + const streamId = events[index]?.streamId ?? '_GET_stream'; + for (const event of events.slice(index + 1).filter(e => e.streamId === streamId)) { + await send(event.id, event.message); + } + return streamId; + } + }; + } + + async function createTransport(options?: { + keepAliveMs?: number; + eventStore?: EventStore; + }): Promise<{ transport: WebStandardStreamableHTTPServerTransport; sessionId: string }> { + const transport = new WebStandardStreamableHTTPServerTransport({ sessionIdGenerator: () => randomUUID(), ...options }); + await new McpServer({ name: 'test-server', version: '1.0.0' }).connect(transport); + const initResponse = await transport.handleRequest(req('POST', { body: TEST_MESSAGES.initialize })); + expect(initResponse.status).toBe(200); + return { transport, sessionId: initResponse.headers.get('mcp-session-id') as string }; + } + + function get(sessionId: string, lastEventId?: string): Request { + const headers: Record = { 'mcp-session-id': sessionId, 'mcp-protocol-version': '2025-11-25' }; + if (lastEventId !== undefined) { + headers['Last-Event-ID'] = lastEventId; + } + return req('GET', { headers }); + } + + /** Opens the standalone GET stream and returns a real stored event id to resume from. */ + async function openGetStreamWithEvent( + transport: WebStandardStreamableHTTPServerTransport, + sessionId: string + ): Promise<{ reader: ReadableStreamDefaultReader; eventId: string }> { + const response = await transport.handleRequest(get(sessionId)); + expect(response.status).toBe(200); + const reader = response.body!.getReader(); + await transport.send({ jsonrpc: '2.0', method: 'notifications/message', params: { level: 'info', data: 'x' } }); + const { value } = await reader.read(); + const eventId = /^id: (.+)$/m.exec(new TextDecoder().decode(value))?.[1]; + expect(eventId).toBeDefined(); + return { reader, eventId: eventId! }; + } + + beforeEach(() => { + vi.useFakeTimers(); + }); + + afterEach(() => { + vi.useRealTimers(); + }); + + it('should keep the resumed stream alive when the predecessor connection is cancelled late', async () => { + const { transport, sessionId } = await createTransport({ eventStore: createSimpleEventStore() }); + const { reader: staleReader, eventId } = await openGetStreamWithEvent(transport, sessionId); + + // Client reconnects and resumes while the old connection is still half-open + const resumed = await transport.handleRequest(get(sessionId, eventId)); + expect(resumed.status).toBe(200); + const resumedReader = resumed.body!.getReader(); + + // The old connection's socket finally dies. This must not tear down the + // resumed stream's keep-alive timer or its stream registration. + await staleReader.cancel(); + + await vi.advanceTimersByTimeAsync(15000); + const { value: frame } = await resumedReader.read(); + expect(new TextDecoder().decode(frame)).toBe(': keepalive\n\n'); + + // Server-to-client messages must still reach the resumed stream too + await transport.send({ jsonrpc: '2.0', method: 'notifications/message', params: { level: 'info', data: 'after' } }); + const { value: notification } = await resumedReader.read(); + expect(new TextDecoder().decode(notification)).toContain('notifications/message'); + + await transport.close(); + expect(vi.getTimerCount()).toBe(0); + }); + + it('should close the predecessor stream when a resume re-registers its stream id', async () => { + const { transport, sessionId } = await createTransport({ eventStore: createSimpleEventStore() }); + const { reader: staleReader, eventId } = await openGetStreamWithEvent(transport, sessionId); + + await transport.handleRequest(get(sessionId, eventId)); + + // The superseded stream must end cleanly rather than hang as a zombie + const { done } = await staleReader.read(); + expect(done).toBe(true); + + 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, sessionId } = await createTransport({ eventStore }); + + // Enter replayEvents and park on the replayEventsAfter await + const pendingGet = transport.handleRequest(get(sessionId, 'evt-1')); + await vi.advanceTimersByTimeAsync(0); + expect(releaseReplay).toBeDefined(); + + // Close the transport mid-await, then let the replay continuation run. + // The deferred continuation must not arm a timer close() can never clear, + // and must not hand out a 200 SSE stream nothing will ever write to. + await transport.close(); + releaseReplay!(); + const response = await pendingGet; + + expect(response.status).toBe(404); + expect(vi.getTimerCount()).toBe(0); + }); + + it('should not leave a keep-alive timer armed when the priming event write fails', async () => { + let failStore = false; + const eventStore: EventStore = { + async storeEvent(streamId: StreamId): Promise { + if (failStore) { + throw new Error('store unavailable'); + } + return `${streamId}#0`; + }, + async replayEventsAfter(): Promise { + return '_GET_stream'; + } + }; + const { transport, sessionId } = await createTransport({ eventStore }); + // Let the init response complete so its own stream cleanup has run + await vi.advanceTimersByTimeAsync(0); + expect(vi.getTimerCount()).toBe(0); + + failStore = true; + const response = await transport.handleRequest( + req('POST', { + body: { jsonrpc: '2.0', method: 'tools/list', params: {}, id: 'req-1' }, + headers: { 'mcp-session-id': sessionId, 'mcp-protocol-version': '2025-11-25' } + }) + ); + + // The 400 discards the Response, so nothing could ever cancel the stream — + // no timer may be left behind. + expect(response.status).toBe(400); + expect(vi.getTimerCount()).toBe(0); + + // The failed request's stream and correlation must be released too: a + // late response for it has nowhere to go and must say so, rather than + // being written to a dead stream. + await expect(transport.send({ jsonrpc: '2.0', id: 'req-1', result: { tools: [] } })).rejects.toThrow( + 'No connection established for request ID' + ); + + await transport.close(); + }); + + it.each([NaN, Infinity])('should disable keep-alive for non-finite keepAliveMs (%s)', async keepAliveMs => { + const { transport, sessionId } = await createTransport({ keepAliveMs }); + + const response = await transport.handleRequest(get(sessionId)); + expect(response.status).toBe(200); + + // A non-finite interval must disable keep-alive, not arm a broken timer + // (setInterval would clamp it to ~1ms and flood the stream). + expect(vi.getTimerCount()).toBe(0); + + await transport.close(); + }); + + it('should clamp keepAliveMs above 2^31-1 instead of flooding the stream', async () => { + const { transport, sessionId } = await createTransport({ keepAliveMs: 2 ** 31 }); + + const response = await transport.handleRequest(get(sessionId)); + const reader = response.body!.getReader(); + + // Un-clamped, setInterval treats 2^31 as ~1ms and floods; clamped, no + // frame is due for a very long time. + await vi.advanceTimersByTimeAsync(60000); + const read = reader.read(); + const raced = await Promise.race([read, Promise.resolve('pending')]); + expect(raced).toBe('pending'); + + await transport.close(); + }); +}); From 8a88ab7c25fff8114a9b32e3830cce7993f4cd69 Mon Sep 17 00:00:00 2001 From: Felix Weinberger Date: Fri, 24 Jul 2026 16:55:36 +0000 Subject: [PATCH 2/6] Handle response delivery races in Streamable HTTP send() A response completing while its SSE stream was being resumed could be written to the evicted stream (lost) or delivered twice when the event store made it replay-visible before the write resolved. A stream that disconnected before its response completed also left the response unstored, so a polling client could never retrieve it, and the request correlation maps leaked. send() now stores response events even when the request stream is disconnected (matching the standalone stream path), re-reads the stream registration after the store write, skips events the resumed stream's replay already delivered, and releases request correlations once the response is safely replayable. Without an event store, or in JSON response mode where a replay can never settle the pending response, completing against a missing stream still surfaces an error. Also guards session initialization and stream registration against a transport that closed while the request body or the session initialization callback was pending. --- src/server/webStandardStreamableHttp.ts | 69 +++- test/server/streamableHttp.test.ts | 421 ++++++++++++++++++++++++ 2 files changed, 483 insertions(+), 7 deletions(-) diff --git a/src/server/webStandardStreamableHttp.ts b/src/server/webStandardStreamableHttp.ts index 531a440c67..97f728f324 100644 --- a/src/server/webStandardStreamableHttp.ts +++ b/src/server/webStandardStreamableHttp.ts @@ -71,6 +71,12 @@ interface StreamMapping { resolveJson?: (response: Response) => void; /** Cleanup function to close stream and remove mapping */ cleanup: () => void; + /** + * Event IDs delivered to this stream by replay when it was registered. + * Lets send() avoid re-delivering events the replay already wrote when a + * resume raced in-flight event-store writes. + */ + replayedEventIds?: Set; } /** @@ -586,6 +592,7 @@ export class WebStandardStreamableHTTPServerTransport implements Transport { }); // Replay events - returns the streamId for backwards compatibility + const replayedEventIds = new Set(); replayedStreamId = await this._eventStore.replayEventsAfter(lastEventId, { send: async (eventId: string, message: JSONRPCMessage) => { const success = this.writeSSEEvent(streamController!, encoder, message, eventId); @@ -596,6 +603,8 @@ export class WebStandardStreamableHTTPServerTransport implements Transport { } catch { // Controller might already be closed } + } else { + replayedEventIds.add(eventId); } } }); @@ -620,6 +629,7 @@ export class WebStandardStreamableHTTPServerTransport implements Transport { this._streamMapping.set(replayedStreamId, { controller: streamController!, encoder, + replayedEventIds, cleanup: () => { if (keepAliveTimer !== undefined) { clearInterval(keepAliveTimer); @@ -748,6 +758,13 @@ export class WebStandardStreamableHTTPServerTransport implements Transport { return this.createJsonErrorResponse(400, -32700, 'Parse error: Invalid JSON-RPC message'); } + // close() may have run while awaiting the request body above: don't + // initialize a session (or fire onsessioninitialized) on a transport + // whose cleanups have already run. + if (this._closed) { + return this.createJsonErrorResponse(404, -32001, 'Session not found'); + } + // Check if this is an initialization request // https://spec.modelcontextprotocol.io/specification/2025-03-26/basic/lifecycle/ const isInitializationRequest = messages.some(isInitializeRequest); @@ -786,6 +803,14 @@ export class WebStandardStreamableHTTPServerTransport implements Transport { } } + // close() may have run while awaiting the request body or the session + // initialization callback above: don't register streams into maps the + // close sweep has already cleared, or hand out a stream that nothing + // will ever write to. + if (this._closed) { + return this.createJsonErrorResponse(404, -32001, 'Session not found'); + } + // check if it contains requests const hasRequests = messages.some(isJSONRPCRequest); @@ -1095,8 +1120,14 @@ export class WebStandardStreamableHTTPServerTransport implements Transport { return; } - // Send the message to the standalone SSE stream - if (standaloneSse.controller && standaloneSse.encoder) { + // Send the message to the standalone SSE stream, unless this stream + // was registered by a replay that already delivered this event (the + // store made it visible before this call resumed). + if ( + standaloneSse.controller && + standaloneSse.encoder && + (eventId === undefined || !standaloneSse.replayedEventIds?.has(eventId)) + ) { this.writeSSEEvent(standaloneSse.controller, standaloneSse.encoder, message, eventId); } return; @@ -1108,17 +1139,27 @@ export class WebStandardStreamableHTTPServerTransport implements Transport { throw new Error(`No connection established for request ID: ${String(requestId)}`); } - const stream = this._streamMapping.get(streamId); + let stream = this._streamMapping.get(streamId); - if (!this._enableJsonResponse && stream?.controller && stream?.encoder) { - // For SSE responses, generate event ID if event store is provided + if (!this._enableJsonResponse) { + // Store the event even if the stream is disconnected (e.g. after + // closeSSEStream switched the client to polling), so it can be + // replayed on reconnect — mirroring the standalone stream path. let eventId: string | undefined; if (this._eventStore) { eventId = await this._eventStore.storeEvent(streamId, message); + // Re-fetch after the await: a resume completing meanwhile replaces + // this stream's registration with a successor, and the message + // must go to the live stream, not the evicted one. + stream = this._streamMapping.get(streamId); + } + // Skip the write if the stream was registered by a replay that + // already delivered this event (the store made it visible before + // this call resumed). + if (stream?.controller && stream?.encoder && (eventId === undefined || !stream.replayedEventIds?.has(eventId))) { + this.writeSSEEvent(stream.controller, stream.encoder, message, eventId); } - // Write the event to the response stream - this.writeSSEEvent(stream.controller, stream.encoder, message, eventId); } if (isJSONRPCResultResponse(message) || isJSONRPCErrorResponse(message)) { @@ -1132,6 +1173,20 @@ export class WebStandardStreamableHTTPServerTransport implements Transport { if (allResponsesReady) { if (!stream) { + if (!this._enableJsonResponse && this._eventStore) { + // The stream went away (client disconnect, or + // closeSSEStream switched the client to polling), but the + // response was stored above and will be replayed on + // resume — release the correlations rather than throwing. + for (const id of relatedIds) { + this._requestResponseMap.delete(id); + this._requestToStreamMapping.delete(id); + } + return; + } + // Without resumability (or in JSON response mode, where the + // pending HTTP response can never be settled by a replay), + // a missing stream means the response cannot be delivered. throw new Error(`No connection established for request ID: ${String(requestId)}`); } if (this._enableJsonResponse && stream.resolveJson) { diff --git a/test/server/streamableHttp.test.ts b/test/server/streamableHttp.test.ts index 83b6e8c30f..584e7fd93f 100644 --- a/test/server/streamableHttp.test.ts +++ b/test/server/streamableHttp.test.ts @@ -3672,4 +3672,425 @@ describe('WebStandardStreamableHTTPServerTransport SSE keep-alive lifecycle', () await transport.close(); }); + + it('should deliver a response to the successor stream when a resume completes during the event-store write', async () => { + // storeEvent parks on the RESPONSE write, so a resume can complete and + // replace the stream registration while send() is awaiting. + const events: { id: string; streamId: string; message: JSONRPCMessage }[] = []; + let counter = 0; + let parkNext = false; + let releaseStore: (() => void) | undefined; + const eventStore: EventStore = { + async storeEvent(streamId: StreamId, message: JSONRPCMessage): Promise { + if (parkNext) { + parkNext = false; + await new Promise(resolve => { + releaseStore = resolve; + }); + } + const id = `${streamId}#${counter++}`; + events.push({ id, streamId, message }); + return id; + }, + async replayEventsAfter(lastEventId: EventId, { send }): Promise { + const index = events.findIndex(e => e.id === lastEventId); + const streamId = events[index]?.streamId ?? '_GET_stream'; + for (const event of events.slice(index + 1).filter(e => e.streamId === streamId)) { + await send(event.id, event.message); + } + return streamId; + } + }; + + const transport = new WebStandardStreamableHTTPServerTransport({ sessionIdGenerator: () => randomUUID(), eventStore }); + const mcpServer = new McpServer({ name: 'test-server', version: '1.0.0' }); + let resolveTool: (() => void) | undefined; + mcpServer.tool('slow', async () => { + await new Promise(resolve => { + resolveTool = resolve; + }); + return { content: [{ type: 'text', text: 'done' }] }; + }); + 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; + + // Original request stream; capture its priming event id for the resume + const original = await transport.handleRequest( + req('POST', { + body: { jsonrpc: '2.0', method: 'tools/call', params: { name: 'slow', arguments: {} }, id: 'call-1' }, + headers: { 'mcp-session-id': sessionId, 'mcp-protocol-version': '2025-11-25' } + }) + ); + const originalReader = original.body!.getReader(); + const { value: priming } = await originalReader.read(); + const primingEventId = /^id: (.+)$/m.exec(new TextDecoder().decode(priming))?.[1]; + expect(primingEventId).toBeDefined(); + + // Complete the tool; the response's storeEvent parks inside send() + parkNext = true; + resolveTool?.(); + await vi.advanceTimersByTimeAsync(0); + expect(releaseStore).toBeDefined(); + + // Client reconnects and resumes the request stream while send() is parked + const resumed = await transport.handleRequest( + req('GET', { headers: { 'mcp-session-id': sessionId, 'mcp-protocol-version': '2025-11-25', 'Last-Event-ID': primingEventId! } }) + ); + expect(resumed.status).toBe(200); + const resumedReader = resumed.body!.getReader(); + + // Release the parked write: the response must reach the live (resumed) + // stream, not vanish into the evicted one. + releaseStore!(); + await vi.advanceTimersByTimeAsync(0); + + let resumedData = ''; + for (let i = 0; i < 3 && !resumedData.includes('call-1'); i++) { + const { value, done } = await resumedReader.read(); + if (done) { + break; + } + resumedData += new TextDecoder().decode(value); + } + expect(resumedData).toContain('"id":"call-1"'); + expect(resumedData).toContain('done'); + + await transport.close(); + expect(vi.getTimerCount()).toBe(0); + }); + + it('should not register streams when the transport closes during the session initialization callback', async () => { + let releaseInit: (() => void) | undefined; + const transport = new WebStandardStreamableHTTPServerTransport({ + sessionIdGenerator: () => randomUUID(), + onsessioninitialized: () => + new Promise(resolve => { + releaseInit = resolve; + }) + }); + await new McpServer({ name: 'test-server', version: '1.0.0' }).connect(transport); + + // Initialization parks on the onsessioninitialized await + const pendingInit = transport.handleRequest(req('POST', { body: TEST_MESSAGES.initialize })); + await vi.advanceTimersByTimeAsync(0); + expect(releaseInit).toBeDefined(); + + // Close mid-await, then let the continuation run: it must not hand out + // a 200 SSE stream nothing will ever write to, nor arm a timer. + await transport.close(); + releaseInit!(); + const response = await pendingInit; + + expect(response.status).toBe(404); + expect(vi.getTimerCount()).toBe(0); + }); + + it('should not double-deliver a response when the event becomes replayable before the store write resolves', async () => { + // storeEvent persists the event (making it replay-visible) and THEN + // parks: a resume in that window replays the response to the successor, + // and the parked send() continuation must not write it a second time. + const events: { id: string; streamId: string; message: JSONRPCMessage }[] = []; + let counter = 0; + let parkNext = false; + let releaseStore: (() => void) | undefined; + const eventStore: EventStore = { + async storeEvent(streamId: StreamId, message: JSONRPCMessage): Promise { + const id = `${streamId}#${counter++}`; + events.push({ id, streamId, message }); + if (parkNext) { + parkNext = false; + await new Promise(resolve => { + releaseStore = resolve; + }); + } + return id; + }, + async replayEventsAfter(lastEventId: EventId, { send }): Promise { + const index = events.findIndex(e => e.id === lastEventId); + const streamId = events[index]?.streamId ?? '_GET_stream'; + for (const event of events.slice(index + 1).filter(e => e.streamId === streamId)) { + await send(event.id, event.message); + } + return streamId; + } + }; + + const transport = new WebStandardStreamableHTTPServerTransport({ sessionIdGenerator: () => randomUUID(), eventStore }); + const mcpServer = new McpServer({ name: 'test-server', version: '1.0.0' }); + let resolveTool: (() => void) | undefined; + mcpServer.tool('slow', async () => { + await new Promise(resolve => { + resolveTool = resolve; + }); + return { content: [{ type: 'text', text: 'done' }] }; + }); + 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; + + const original = await transport.handleRequest( + req('POST', { + body: { jsonrpc: '2.0', method: 'tools/call', params: { name: 'slow', arguments: {} }, id: 'call-1' }, + headers: { 'mcp-session-id': sessionId, 'mcp-protocol-version': '2025-11-25' } + }) + ); + const originalReader = original.body!.getReader(); + const { value: priming } = await originalReader.read(); + const primingEventId = /^id: (.+)$/m.exec(new TextDecoder().decode(priming))?.[1]; + expect(primingEventId).toBeDefined(); + + parkNext = true; + resolveTool?.(); + await vi.advanceTimersByTimeAsync(0); + expect(releaseStore).toBeDefined(); + + // The response event is already visible in the store: the resume's + // replay delivers it to the successor stream. + const resumed = await transport.handleRequest( + req('GET', { headers: { 'mcp-session-id': sessionId, 'mcp-protocol-version': '2025-11-25', 'Last-Event-ID': primingEventId! } }) + ); + const resumedReader = resumed.body!.getReader(); + + releaseStore!(); + await vi.advanceTimersByTimeAsync(0); + + let resumedData = ''; + for (let i = 0; i < 4; i++) { + const { value, done } = await resumedReader.read(); + if (done) { + break; + } + resumedData += new TextDecoder().decode(value); + } + const deliveries = resumedData.match(/"id":"call-1"/g) ?? []; + expect(deliveries).toHaveLength(1); + + await transport.close(); + expect(vi.getTimerCount()).toBe(0); + }); + + it('should not initialize a session when the transport closes while the request body is being read', async () => { + const onsessioninitialized = vi.fn(); + const transport = new WebStandardStreamableHTTPServerTransport({ + sessionIdGenerator: () => randomUUID(), + onsessioninitialized + }); + await new McpServer({ name: 'test-server', version: '1.0.0' }).connect(transport); + + // Init request whose body stream parks until we release it + let releaseBody: (() => void) | undefined; + const encoder = new TextEncoder(); + const body = new ReadableStream({ + start: controller => { + releaseBody = () => { + controller.enqueue(encoder.encode(JSON.stringify(TEST_MESSAGES.initialize))); + controller.close(); + }; + } + }); + const request = new Request('http://localhost/mcp', { + method: 'POST', + headers: { Accept: 'application/json, text/event-stream', 'Content-Type': 'application/json' }, + body, + // @ts-expect-error duplex is required for streaming bodies but not yet in lib types + duplex: 'half' + }); + + const pending = transport.handleRequest(request); + await vi.advanceTimersByTimeAsync(0); + expect(releaseBody).toBeDefined(); + + // Close while req.json() is still reading, then deliver the body: the + // continuation must not initialize a session on the closed transport. + await transport.close(); + releaseBody!(); + const response = await pending; + + expect(response.status).toBe(404); + expect(onsessioninitialized).not.toHaveBeenCalled(); + expect(vi.getTimerCount()).toBe(0); + }); + + it('should store and replay a response completed after closeSSEStream switched the client to polling', async () => { + const events: { id: string; streamId: string; message: JSONRPCMessage }[] = []; + let counter = 0; + const eventStore: EventStore = { + async storeEvent(streamId: StreamId, message: JSONRPCMessage): Promise { + const id = `${streamId}#${counter++}`; + events.push({ id, streamId, message }); + return id; + }, + async replayEventsAfter(lastEventId: EventId, { send }): Promise { + const index = events.findIndex(e => e.id === lastEventId); + const streamId = events[index]?.streamId ?? '_GET_stream'; + for (const event of events.slice(index + 1).filter(e => e.streamId === streamId)) { + await send(event.id, event.message); + } + return streamId; + } + }; + const transport = new WebStandardStreamableHTTPServerTransport({ sessionIdGenerator: () => randomUUID(), eventStore }); + const mcpServer = new McpServer({ name: 'test-server', version: '1.0.0' }); + let resolveTool: (() => void) | undefined; + mcpServer.tool('slow', async () => { + await new Promise(resolve => { + resolveTool = resolve; + }); + return { content: [{ type: 'text', text: 'done' }] }; + }); + 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; + + const original = await transport.handleRequest( + req('POST', { + body: { jsonrpc: '2.0', method: 'tools/call', params: { name: 'slow', arguments: {} }, id: 'call-1' }, + headers: { 'mcp-session-id': sessionId, 'mcp-protocol-version': '2025-11-25' } + }) + ); + const { value: priming } = await original.body!.getReader().read(); + const primingEventId = /^id: (.+)$/m.exec(new TextDecoder().decode(priming))?.[1]; + expect(primingEventId).toBeDefined(); + + // Server switches the client to polling; the request stream is gone + transport.closeSSEStream('call-1'); + + // Tool completes with no stream attached: the response must be stored + // for replay, and the released correlations must not make send() throw. + resolveTool?.(); + await vi.advanceTimersByTimeAsync(0); + + // Client polls back in with Last-Event-ID and must receive the response + const resumed = await transport.handleRequest( + req('GET', { headers: { 'mcp-session-id': sessionId, 'mcp-protocol-version': '2025-11-25', 'Last-Event-ID': primingEventId! } }) + ); + expect(resumed.status).toBe(200); + const resumedReader = resumed.body!.getReader(); + let resumedData = ''; + for (let i = 0; i < 3 && !resumedData.includes('call-1'); i++) { + const { value, done } = await resumedReader.read(); + if (done) { + break; + } + resumedData += new TextDecoder().decode(value); + } + expect(resumedData).toContain('"id":"call-1"'); + expect(resumedData).toContain('done'); + + await transport.close(); + expect(vi.getTimerCount()).toBe(0); + }); + + it('should not re-deliver replayed server notifications to a resumed standalone stream', async () => { + const events: { id: string; streamId: string; message: JSONRPCMessage }[] = []; + let counter = 0; + let parkNext = false; + let releaseStore: (() => void) | undefined; + const eventStore: EventStore = { + async storeEvent(streamId: StreamId, message: JSONRPCMessage): Promise { + const id = `${streamId}#${counter++}`; + events.push({ id, streamId, message }); + if (parkNext) { + parkNext = false; + await new Promise(resolve => { + releaseStore = resolve; + }); + } + return id; + }, + async replayEventsAfter(lastEventId: EventId, { send }): Promise { + const index = events.findIndex(e => e.id === lastEventId); + const streamId = events[index]?.streamId ?? '_GET_stream'; + for (const event of events.slice(index + 1).filter(e => e.streamId === streamId)) { + await send(event.id, event.message); + } + return streamId; + } + }; + const { transport, sessionId } = await createTransport({ eventStore }); + + // First notification anchors the resume point + const first = await transport.handleRequest(get(sessionId)); + const firstReader = first.body!.getReader(); + await transport.send({ jsonrpc: '2.0', method: 'notifications/message', params: { level: 'info', data: 'anchor' } }); + const { value } = await firstReader.read(); + const anchorId = /^id: (.+)$/m.exec(new TextDecoder().decode(value))?.[1]; + expect(anchorId).toBeDefined(); + + // Second notification: stored (replay-visible), then the write parks + parkNext = true; + const parkedSend = transport.send({ jsonrpc: '2.0', method: 'notifications/message', params: { level: 'info', data: 'raced' } }); + await vi.advanceTimersByTimeAsync(0); + expect(releaseStore).toBeDefined(); + + // The old connection drops and the client resumes: replay delivers the + // second notification to the successor stream. + await firstReader.cancel(); + const resumed = await transport.handleRequest(get(sessionId, anchorId)); + const resumedReader = resumed.body!.getReader(); + + releaseStore!(); + await parkedSend; + + let resumedData = ''; + for (let i = 0; i < 3; i++) { + const read = resumedReader.read(); + const raced = await Promise.race([read, Promise.resolve('pending')]); + if (raced === 'pending') { + break; + } + const { value: chunk, done } = raced as ReadableStreamReadResult; + if (done) { + break; + } + resumedData += new TextDecoder().decode(chunk); + } + const deliveries = resumedData.match(/raced/g) ?? []; + expect(deliveries).toHaveLength(1); + + await transport.close(); + }); + + it('should fail loudly when a JSON-mode response completes after its stream is gone', async () => { + const transport = new WebStandardStreamableHTTPServerTransport({ + sessionIdGenerator: () => randomUUID(), + enableJsonResponse: true + }); + const mcpServer = new McpServer({ name: 'test-server', version: '1.0.0' }); + let resolveTool: (() => void) | undefined; + mcpServer.tool('slow', async () => { + await new Promise(resolve => { + resolveTool = resolve; + }); + return { content: [{ type: 'text', text: 'done' }] }; + }); + 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; + + const errors: Error[] = []; + mcpServer.server.onerror = error => { + errors.push(error); + }; + + void transport.handleRequest( + req('POST', { + body: { jsonrpc: '2.0', method: 'tools/call', params: { name: 'slow', arguments: {} }, id: 'call-1' }, + headers: { 'mcp-session-id': sessionId, 'mcp-protocol-version': '2025-11-25' } + }) + ); + await vi.advanceTimersByTimeAsync(0); + + // The stream mapping disappears while the tool is still running + transport.closeSSEStream('call-1'); + resolveTool?.(); + await vi.advanceTimersByTimeAsync(0); + + // In JSON mode nothing can ever replay the response: completing against + // a missing stream must surface an error, not vanish silently. + expect(errors.map(e => e.message).join('\n')).toContain('No connection established for request ID'); + + await transport.close(); + }); }); From 0f4fb3768ed4709d7aef05c75087edb2e10a9fe3 Mon Sep 17 00:00:00 2001 From: Felix Weinberger Date: Mon, 27 Jul 2026 14:05:06 +0000 Subject: [PATCH 3/6] Align keep-alive implementation with the main branch Extract keep-alive validation into a shared sseKeepAlive helper matching main: non-finite and sub-millisecond intervals disable keep-alive, and oversized delays are clamped instead of firing every millisecond. Also ported from the main-branch implementation: handleRequest returns 404 once the transport is closed instead of registering streams nothing can clean up, SSE responses send X-Accel-Buffering: no (and the POST stream gains no-transform) so proxy buffering does not swallow frames, and a DELETE request closes the transport even when the onsessionclosed callback throws. send() now also releases request correlations before reporting an undeliverable response, so the entries don't outlive it. --- src/server/sseKeepAlive.ts | 15 +++++ src/server/webStandardStreamableHttp.ts | 74 ++++++++++++------------- test/server/sseKeepAlive.test.ts | 29 ++++++++++ test/server/streamableHttp.test.ts | 40 +++++++++++++ 4 files changed, 120 insertions(+), 38 deletions(-) create mode 100644 src/server/sseKeepAlive.ts create mode 100644 test/server/sseKeepAlive.test.ts diff --git a/src/server/sseKeepAlive.ts b/src/server/sseKeepAlive.ts new file mode 100644 index 0000000000..25df11871c --- /dev/null +++ b/src/server/sseKeepAlive.ts @@ -0,0 +1,15 @@ +/** Default interval between SSE keep-alive comment frames. */ +export const DEFAULT_SSE_KEEP_ALIVE_MS = 15_000; + +const MAX_TIMER_DELAY_MS = 2 ** 31 - 1; + +/** Arms an unref'd timer, or disables keep-alive for invalid delays. */ +export function armSseKeepAlive(intervalMs: number, onTick: () => void): ReturnType | undefined { + if (!Number.isFinite(intervalMs) || intervalMs < 1) { + return undefined; + } + + const timer = setInterval(onTick, Math.min(intervalMs, MAX_TIMER_DELAY_MS)); + (timer as { unref?: () => void }).unref?.(); + return timer; +} diff --git a/src/server/webStandardStreamableHttp.ts b/src/server/webStandardStreamableHttp.ts index 97f728f324..f91280170b 100644 --- a/src/server/webStandardStreamableHttp.ts +++ b/src/server/webStandardStreamableHttp.ts @@ -10,6 +10,7 @@ import { isJsonContentType } from '../shared/mediaType.js'; import { Transport } from '../shared/transport.js'; import { AuthInfo } from './auth/types.js'; +import { armSseKeepAlive, DEFAULT_SSE_KEEP_ALIVE_MS } from './sseKeepAlive.js'; import { MessageExtraInfo, RequestInfo, @@ -154,23 +155,12 @@ export interface WebStandardStreamableHTTPServerTransportOptions { retryInterval?: number; /** - * Interval in milliseconds between SSE keep-alive comment frames (`: keepalive`) - * written to open SSE streams. Keep-alive frames prevent idle streams (e.g. the - * standalone GET stream, or a POST stream during a long-running tool call) from - * being killed by intermediaries and server idle timeouts, which clients observe - * as `SSE stream disconnected: TypeError: terminated`. - * - * 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. Non-finite values also - * disable; intervals above 2^31-1 ms are clamped to that maximum. + * Interval in milliseconds between SSE keep-alive comment frames. + * Defaults to `15000`; values below 1 (including `0`) disable keep-alive. */ keepAliveMs?: number; } -/** Default interval between SSE keep-alive comment frames. */ -const DEFAULT_KEEP_ALIVE_MS = 15_000; - /** * Options for handling a request */ @@ -266,7 +256,7 @@ export class WebStandardStreamableHTTPServerTransport implements Transport { this._allowedOrigins = options.allowedOrigins; this._enableDnsRebindingProtection = options.enableDnsRebindingProtection ?? false; this._retryInterval = options.retryInterval; - this._keepAliveMs = options.keepAliveMs ?? DEFAULT_KEEP_ALIVE_MS; + this._keepAliveMs = options.keepAliveMs ?? DEFAULT_SSE_KEEP_ALIVE_MS; } /** @@ -280,24 +270,16 @@ export class WebStandardStreamableHTTPServerTransport implements Transport { controller: ReadableStreamDefaultController, encoder: TextEncoder ): ReturnType | undefined { - // Non-finite values disable keep-alive rather than arming a broken timer, - // and delays above 2^31-1 are clamped (setInterval treats them as 1ms). // After close() nothing may arm: the stream cleanups have already run. - if (!Number.isFinite(this._keepAliveMs) || this._keepAliveMs <= 0 || this._closed) { - return undefined; - } - const timer = setInterval( - () => { - try { - controller.enqueue(encoder.encode(': keepalive\n\n')); - } catch { - clearInterval(timer); - } - }, - Math.min(this._keepAliveMs, 2 ** 31 - 1) - ); - // Don't let the keep-alive timer hold the process open (Node.js only) - (timer as { unref?: () => void }).unref?.(); + if (this._closed) return undefined; + + const timer = armSseKeepAlive(this._keepAliveMs, () => { + try { + controller.enqueue(encoder.encode(': keepalive\n\n')); + } catch { + if (timer !== undefined) clearInterval(timer); + } + }); return timer; } @@ -379,6 +361,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) { @@ -498,7 +484,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 @@ -559,7 +546,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) { @@ -880,8 +868,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 @@ -978,9 +967,12 @@ export class WebStandardStreamableHTTPServerTransport implements Transport { return protocolError; } - await Promise.resolve(this._onsessionclosed?.(this.sessionId!)); - await this.close(); - return new Response(null, { status: 200 }); + try { + await Promise.resolve(this._onsessionclosed?.(this.sessionId!)); + return new Response(null, { status: 200 }); + } finally { + await this.close(); + } } /** @@ -1187,6 +1179,12 @@ export class WebStandardStreamableHTTPServerTransport implements Transport { // Without resumability (or in JSON response mode, where the // pending HTTP response can never be settled by a replay), // a missing stream means the response cannot be delivered. + // Release the correlations so the entries don't outlive the + // undeliverable response. + for (const id of relatedIds) { + this._requestResponseMap.delete(id); + this._requestToStreamMapping.delete(id); + } throw new Error(`No connection established for request ID: ${String(requestId)}`); } if (this._enableJsonResponse && stream.resolveJson) { diff --git a/test/server/sseKeepAlive.test.ts b/test/server/sseKeepAlive.test.ts new file mode 100644 index 0000000000..238ac2f81c --- /dev/null +++ b/test/server/sseKeepAlive.test.ts @@ -0,0 +1,29 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; + +import { armSseKeepAlive } from '../../src/server/sseKeepAlive.js'; + +describe('armSseKeepAlive', () => { + beforeEach(() => vi.useFakeTimers()); + afterEach(() => vi.useRealTimers()); + + it.each([0, -1, 0.5, Number.NaN, Number.POSITIVE_INFINITY])('disables invalid delay %s', delay => { + expect(armSseKeepAlive(delay, () => {})).toBeUndefined(); + expect(vi.getTimerCount()).toBe(0); + }); + + it('ticks at the configured interval', async () => { + const tick = vi.fn(); + const timer = armSseKeepAlive(1_000, tick)!; + await vi.advanceTimersByTimeAsync(3_000); + expect(tick).toHaveBeenCalledTimes(3); + clearInterval(timer); + }); + + it('clamps overflowing delays instead of creating a 1ms timer', async () => { + const tick = vi.fn(); + const timer = armSseKeepAlive(2 ** 31, tick)!; + await vi.advanceTimersByTimeAsync(60_000); + expect(tick).not.toHaveBeenCalled(); + clearInterval(timer); + }); +}); diff --git a/test/server/streamableHttp.test.ts b/test/server/streamableHttp.test.ts index 584e7fd93f..de3c2f4df7 100644 --- a/test/server/streamableHttp.test.ts +++ b/test/server/streamableHttp.test.ts @@ -4093,4 +4093,44 @@ describe('WebStandardStreamableHTTPServerTransport SSE keep-alive lifecycle', () await transport.close(); }); + + it('should close the transport when the onsessionclosed callback throws on DELETE', async () => { + const transport = new WebStandardStreamableHTTPServerTransport({ + sessionIdGenerator: () => randomUUID(), + onsessionclosed: () => { + throw new Error('session registry unavailable'); + } + }); + 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; + + // Open the standalone stream so a keep-alive timer is armed + const getResponse = await transport.handleRequest(get(sessionId)); + expect(getResponse.status).toBe(200); + expect(vi.getTimerCount()).toBe(1); + + await expect( + transport.handleRequest( + new Request('http://localhost/mcp', { + method: 'DELETE', + headers: { 'mcp-session-id': sessionId, 'mcp-protocol-version': '2025-11-25' } + }) + ) + ).rejects.toThrow('session registry unavailable'); + + // The callback threw, but the transport must still have been closed: + // timers swept and subsequent requests rejected. + expect(vi.getTimerCount()).toBe(0); + const after = await transport.handleRequest(get(sessionId)); + expect(after.status).toBe(404); + }); + + it('should reject requests with 404 after the transport is closed', async () => { + const { transport, sessionId } = await createTransport(); + await transport.close(); + + const response = await transport.handleRequest(get(sessionId)); + expect(response.status).toBe(404); + }); }); From 56cf4c1c7d29a4b2809e98968683121f8a7b3f7e Mon Sep 17 00:00:00 2001 From: Felix Weinberger Date: Mon, 27 Jul 2026 14:41:40 +0000 Subject: [PATCH 4/6] Gate the replay hand-off in send() on a delivered cursor send() released a completed response's correlations to Last-Event-ID replay whenever an event store was configured, but a client can only resume a stream if it actually holds a cursor: clients that never received an id-bearing event (for example pre-2025-11-25 clients whose stream carried no notifications) can never issue the resume, so their lost responses were released silently where they previously surfaced through onerror. Request streams are now marked resumable when an id-bearing event is delivered on them - the priming event, a stored notification, or a replay - and the silent hand-off applies only to those streams. A response completing while the transport closes is a no-op instead of a spurious error, failed POST registrations release their tracking entry, and the changeset now also describes the send() delivery-model changes. --- .changeset/sse-keepalive-timer-lifecycle.md | 3 +- src/server/webStandardStreamableHttp.ts | 28 +++- test/server/streamableHttp.test.ts | 177 ++++++++++++++++++++ 3 files changed, 205 insertions(+), 3 deletions(-) diff --git a/.changeset/sse-keepalive-timer-lifecycle.md b/.changeset/sse-keepalive-timer-lifecycle.md index a0523949fd..030f1ec4fe 100644 --- a/.changeset/sse-keepalive-timer-lifecycle.md +++ b/.changeset/sse-keepalive-timer-lifecycle.md @@ -4,4 +4,5 @@ Fix SSE keep-alive timer lifecycle in the Streamable HTTP server transport. Keep-alive timers are now owned by the stream they were armed for, so a stale disconnect arriving after a reconnect no longer stops the live stream's keep-alive, a resume closes the superseded stream cleanly, and a failed priming-event write no longer leaves a timer running or a stale request correlation behind. A resume that completes after the transport closed now gets a 404 instead of a silent stream nothing will ever write to. Non-finite `keepAliveMs` values now disable -keep-alive, and values above 2^31-1 are clamped instead of firing every millisecond. +keep-alive, and values above 2^31-1 are clamped instead of firing every millisecond. With an event store configured, a response completing while its request stream is disconnected is now stored for `Last-Event-ID` replay and its request correlations released when the client can +resume the stream, instead of `send()` rejecting; for clients that cannot resume (no priming event), `send()` still rejects so the failure surfaces. Replays no longer double-deliver events written concurrently with a resume, and requests racing transport close now receive a 404. diff --git a/src/server/webStandardStreamableHttp.ts b/src/server/webStandardStreamableHttp.ts index f91280170b..b83d5fc4db 100644 --- a/src/server/webStandardStreamableHttp.ts +++ b/src/server/webStandardStreamableHttp.ts @@ -227,6 +227,12 @@ export class WebStandardStreamableHTTPServerTransport implements Transport { private _hasHandledRequest: boolean = false; private _streamMapping: Map = new Map(); private _requestToStreamMapping: Map = new Map(); + /** + * Request streams whose client holds a Last-Event-ID cursor (at least one + * id-bearing event was delivered on them), so a stored response can be + * replayed on resume. + */ + private _resumableStreams: Set = new Set(); private _requestResponseMap: Map = new Map(); private _initialized: boolean = false; private _enableJsonResponse: boolean = false; @@ -419,6 +425,7 @@ export class WebStandardStreamableHTTPServerTransport implements Transport { primingEvent = `id: ${primingEventId}\nretry: ${this._retryInterval}\ndata: \n\n`; } controller.enqueue(encoder.encode(primingEvent)); + this._resumableStreams.add(streamId); } /** @@ -631,6 +638,9 @@ export class WebStandardStreamableHTTPServerTransport implements Transport { } }); + // The resume itself proves the client holds a Last-Event-ID cursor + this._resumableStreams.add(replayedStreamId); + keepAliveTimer = this.startKeepAlive(streamController!, encoder); return new Response(readable, { headers }); @@ -928,6 +938,7 @@ export class WebStandardStreamableHTTPServerTransport implements Transport { // Failing before the Response is handed out means nothing can ever // cancel this stream: release it and its request correlations here. this._streamMapping.get(streamId)?.cleanup(); + this._resumableStreams.delete(streamId); for (const message of messages) { if (isJSONRPCRequest(message)) { this._requestToStreamMapping.delete(message.id); @@ -1053,6 +1064,7 @@ export class WebStandardStreamableHTTPServerTransport implements Transport { // Clear any pending responses this._requestResponseMap.clear(); + this._resumableStreams.clear(); this.onclose?.(); } @@ -1150,7 +1162,11 @@ export class WebStandardStreamableHTTPServerTransport implements Transport { // already delivered this event (the store made it visible before // this call resumed). if (stream?.controller && stream?.encoder && (eventId === undefined || !stream.replayedEventIds?.has(eventId))) { - this.writeSSEEvent(stream.controller, stream.encoder, message, eventId); + const written = this.writeSSEEvent(stream.controller, stream.encoder, message, eventId); + // The client now holds a Last-Event-ID cursor for this stream + if (written && eventId !== undefined) { + this._resumableStreams.add(streamId); + } } } @@ -1165,7 +1181,13 @@ export class WebStandardStreamableHTTPServerTransport implements Transport { if (allResponsesReady) { if (!stream) { - if (!this._enableJsonResponse && this._eventStore) { + // close() ran while the event store write above was awaiting: + // every map was already swept, nothing remains to release or + // report. + if (this._closed) { + return; + } + if (!this._enableJsonResponse && this._eventStore && this._resumableStreams.has(streamId)) { // The stream went away (client disconnect, or // closeSSEStream switched the client to polling), but the // response was stored above and will be replayed on @@ -1174,6 +1196,7 @@ export class WebStandardStreamableHTTPServerTransport implements Transport { this._requestResponseMap.delete(id); this._requestToStreamMapping.delete(id); } + this._resumableStreams.delete(streamId); return; } // Without resumability (or in JSON response mode, where the @@ -1212,6 +1235,7 @@ export class WebStandardStreamableHTTPServerTransport implements Transport { this._requestResponseMap.delete(id); this._requestToStreamMapping.delete(id); } + this._resumableStreams.delete(streamId); } } } diff --git a/test/server/streamableHttp.test.ts b/test/server/streamableHttp.test.ts index de3c2f4df7..4842b58c53 100644 --- a/test/server/streamableHttp.test.ts +++ b/test/server/streamableHttp.test.ts @@ -4133,4 +4133,181 @@ describe('WebStandardStreamableHTTPServerTransport SSE keep-alive lifecycle', () const response = await transport.handleRequest(get(sessionId)); expect(response.status).toBe(404); }); + + it('should surface an error when a response completes for a disconnected client that cannot resume', async () => { + // Pre-2025-11-25 clients never receive a priming event, so they cannot + // resume a request stream: a response completing after their disconnect + // is undeliverable and must surface, not be silently handed to replay. + const eventStore = createSimpleEventStore(); + const transport = new WebStandardStreamableHTTPServerTransport({ sessionIdGenerator: () => randomUUID(), eventStore }); + const mcpServer = new McpServer({ name: 'test-server', version: '1.0.0' }); + let resolveTool: (() => void) | undefined; + mcpServer.tool('slow', async () => { + await new Promise(resolve => { + resolveTool = resolve; + }); + return { content: [{ type: 'text', text: 'done' }] }; + }); + await mcpServer.connect(transport); + const initResponse = await transport.handleRequest( + req('POST', { + body: { + jsonrpc: '2.0', + id: 1, + method: 'initialize', + params: { protocolVersion: '2025-03-26', capabilities: {}, clientInfo: { name: 'legacy', version: '1.0' } } + } + }) + ); + const sessionId = initResponse.headers.get('mcp-session-id') as string; + + const errors: Error[] = []; + mcpServer.server.onerror = error => { + errors.push(error); + }; + + const response = await transport.handleRequest( + req('POST', { + body: { jsonrpc: '2.0', method: 'tools/call', params: { name: 'slow', arguments: {} }, id: 'legacy-1' }, + headers: { 'mcp-session-id': sessionId, 'mcp-protocol-version': '2025-03-26' } + }) + ); + expect(response.status).toBe(200); + + // Client disconnects mid-call; no priming event was ever written, so + // no Last-Event-ID cursor exists for a resume. + await response.body!.getReader().cancel(); + resolveTool?.(); + await vi.advanceTimersByTimeAsync(0); + + expect(errors.map(e => e.message).join('\n')).toContain('No connection established for request ID'); + + await transport.close(); + }); + + it('should hand off to replay for a legacy client that received an id-bearing notification', async () => { + // A pre-2025-11-25 client gets no priming event, but any stored + // notification delivered on the stream carries an id — that cursor is + // enough to resume, so the completed response must be released to + // replay, not surfaced as an error. + const eventStore = createSimpleEventStore(); + const transport = new WebStandardStreamableHTTPServerTransport({ sessionIdGenerator: () => randomUUID(), eventStore }); + const mcpServer = new McpServer({ name: 'test-server', version: '1.0.0' }, { capabilities: { logging: {} } }); + let resolveTool: (() => void) | undefined; + mcpServer.tool('notifying', async (extra): Promise => { + await extra.sendNotification({ method: 'notifications/message', params: { level: 'info', data: 'progress' } }); + await new Promise(resolve => { + resolveTool = resolve; + }); + return { content: [{ type: 'text', text: 'done' }] }; + }); + await mcpServer.connect(transport); + const initResponse = await transport.handleRequest( + req('POST', { + body: { + jsonrpc: '2.0', + id: 1, + method: 'initialize', + params: { protocolVersion: '2025-03-26', capabilities: {}, clientInfo: { name: 'legacy', version: '1.0' } } + } + }) + ); + const sessionId = initResponse.headers.get('mcp-session-id') as string; + + const errors: Error[] = []; + mcpServer.server.onerror = error => { + errors.push(error); + }; + + const response = await transport.handleRequest( + req('POST', { + body: { jsonrpc: '2.0', method: 'tools/call', params: { name: 'notifying', arguments: {} }, id: 'legacy-2' }, + headers: { 'mcp-session-id': sessionId, 'mcp-protocol-version': '2025-03-26' } + }) + ); + const reader = response.body!.getReader(); + await vi.advanceTimersByTimeAsync(0); + const { value } = await reader.read(); + const cursor = /^id: (.+)$/m.exec(new TextDecoder().decode(value))?.[1]; + expect(cursor).toBeDefined(); + + // Client disconnects holding the notification's event id, then the tool completes + await reader.cancel(); + resolveTool?.(); + await vi.advanceTimersByTimeAsync(0); + + expect(errors).toHaveLength(0); + + // The resume from that cursor must replay the response + const resumed = await transport.handleRequest(get(sessionId, cursor!)); + const resumedReader = resumed.body!.getReader(); + let resumedData = ''; + for (let i = 0; i < 3 && !resumedData.includes('legacy-2'); i++) { + const { value: chunk, done } = await resumedReader.read(); + if (done) { + break; + } + resumedData += new TextDecoder().decode(chunk); + } + expect(resumedData).toContain('"id":"legacy-2"'); + + await transport.close(); + }); + + it('should not surface an error when the transport closes during the response store write', async () => { + let parkNext = false; + let releaseStore: (() => void) | undefined; + const inner = createSimpleEventStore(); + const eventStore: EventStore = { + async storeEvent(streamId: StreamId, message: JSONRPCMessage): Promise { + if (parkNext) { + parkNext = false; + await new Promise(resolve => { + releaseStore = resolve; + }); + } + return inner.storeEvent(streamId, message); + }, + replayEventsAfter: inner.replayEventsAfter.bind(inner) + }; + const transport = new WebStandardStreamableHTTPServerTransport({ sessionIdGenerator: () => randomUUID(), eventStore }); + const mcpServer = new McpServer({ name: 'test-server', version: '1.0.0' }); + let resolveTool: (() => void) | undefined; + mcpServer.tool('slow', async () => { + await new Promise(resolve => { + resolveTool = resolve; + }); + return { content: [{ type: 'text', text: 'done' }] }; + }); + 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; + + const errors: Error[] = []; + mcpServer.server.onerror = error => { + errors.push(error); + }; + + void transport.handleRequest( + req('POST', { + body: { jsonrpc: '2.0', method: 'tools/call', params: { name: 'slow', arguments: {} }, id: 'race-1' }, + headers: { 'mcp-session-id': sessionId, 'mcp-protocol-version': '2025-11-25' } + }) + ); + await vi.advanceTimersByTimeAsync(0); + + // The response send parks inside storeEvent; close() sweeps everything + parkNext = true; + resolveTool?.(); + await vi.advanceTimersByTimeAsync(0); + expect(releaseStore).toBeDefined(); + await transport.close(); + releaseStore!(); + await vi.advanceTimersByTimeAsync(0); + + // The transport is gone; the late completion must be a no-op, not a + // spurious 'No connection established' error. + expect(errors.filter(e => e.message.includes('No connection established'))).toHaveLength(0); + expect(vi.getTimerCount()).toBe(0); + }); }); From 0bd3cab37fa8ec189e31a50dc4232035db2d5fd2 Mon Sep 17 00:00:00 2001 From: Felix Weinberger Date: Mon, 27 Jul 2026 15:37:52 +0000 Subject: [PATCH 5/6] Release re-created correlations when a response completes after close The closed-transport no-op in send() ran after the response was recorded in _requestResponseMap, so a send parked on the event-store write when close() swept the maps re-populated them and stranded the response payload and its correlations on the closed transport. Release them before returning. --- src/server/webStandardStreamableHttp.ts | 8 ++++++-- 1 file changed, 6 insertions(+), 2 deletions(-) diff --git a/src/server/webStandardStreamableHttp.ts b/src/server/webStandardStreamableHttp.ts index b83d5fc4db..fe3ba3fa8c 100644 --- a/src/server/webStandardStreamableHttp.ts +++ b/src/server/webStandardStreamableHttp.ts @@ -1182,9 +1182,13 @@ export class WebStandardStreamableHTTPServerTransport implements Transport { if (allResponsesReady) { if (!stream) { // close() ran while the event store write above was awaiting: - // every map was already swept, nothing remains to release or - // report. + // release the entries this call re-created after the sweep, + // then treat the late completion as a no-op. if (this._closed) { + for (const id of relatedIds) { + this._requestResponseMap.delete(id); + this._requestToStreamMapping.delete(id); + } return; } if (!this._enableJsonResponse && this._eventStore && this._resumableStreams.has(streamId)) { From 0f983d0172cf277ebdab85e5775d092281e52369 Mon Sep 17 00:00:00 2001 From: Felix Weinberger Date: Mon, 27 Jul 2026 16:18:49 +0000 Subject: [PATCH 6/6] Make close() idempotent Repeated close() re-ran the stream sweep and re-fired onclose. The main branch guards close() re-entry; this restores the same guard, which the DELETE handler's unconditional close in its finally block now relies on. --- src/server/webStandardStreamableHttp.ts | 3 +++ test/server/streamableHttp.test.ts | 13 +++++++++++++ 2 files changed, 16 insertions(+) diff --git a/src/server/webStandardStreamableHttp.ts b/src/server/webStandardStreamableHttp.ts index fe3ba3fa8c..db83c7cf22 100644 --- a/src/server/webStandardStreamableHttp.ts +++ b/src/server/webStandardStreamableHttp.ts @@ -1052,6 +1052,9 @@ export class WebStandardStreamableHTTPServerTransport implements Transport { } async close(): Promise { + if (this._closed) { + return; + } // Stop any deferred keep-alive arm (e.g. a replay await resolving after // close) from creating a timer nothing can clear. this._closed = true; diff --git a/test/server/streamableHttp.test.ts b/test/server/streamableHttp.test.ts index 4842b58c53..99a6952088 100644 --- a/test/server/streamableHttp.test.ts +++ b/test/server/streamableHttp.test.ts @@ -4310,4 +4310,17 @@ describe('WebStandardStreamableHTTPServerTransport SSE keep-alive lifecycle', () expect(errors.filter(e => e.message.includes('No connection established'))).toHaveLength(0); expect(vi.getTimerCount()).toBe(0); }); + + it('should make close() idempotent', async () => { + const { transport } = await createTransport(); + let oncloseCalls = 0; + transport.onclose = () => { + oncloseCalls++; + }; + + await transport.close(); + await transport.close(); + + expect(oncloseCalls).toBe(1); + }); });