From 916f38a1b86ebabd653109f358aa26154bf8787e Mon Sep 17 00:00:00 2001 From: Sanath Vengaldas Date: Mon, 26 Oct 2026 13:16:04 +0000 Subject: [PATCH 1/3] fix(server): settle the POST for a cancelled request A cancelled request kept its stream mapping forever because Protocol drops the result and the error for an aborted request, so send() never runs and the transport never resolves the POST or closes the per-request SSE stream. Settle the request in the transport when notifications/cancelled arrives: retire both mappings and either resolve the JSON response with the existing empty 202 or clean up the SSE stream. An unresponded sibling that shares the stream keeps it open. --- packages/server/src/server/streamableHttp.ts | 81 +++++++++++- .../server/test/server/streamableHttp.test.ts | 117 ++++++++++++++++++ 2 files changed, 194 insertions(+), 4 deletions(-) diff --git a/packages/server/src/server/streamableHttp.ts b/packages/server/src/server/streamableHttp.ts index 2484ae19b0..8bebc53b17 100644 --- a/packages/server/src/server/streamableHttp.ts +++ b/packages/server/src/server/streamableHttp.ts @@ -7,12 +7,21 @@ * For Node.js Express/HTTP compatibility, use {@linkcode @modelcontextprotocol/node!NodeStreamableHTTPServerTransport | NodeStreamableHTTPServerTransport} which wraps this transport. */ -import type { AuthInfo, JSONRPCMessage, JSONRPCRequest, MessageExtraInfo, RequestId, Transport } from '@modelcontextprotocol/core-internal'; +import type { + AuthInfo, + CancelledNotificationParams, + JSONRPCMessage, + JSONRPCRequest, + MessageExtraInfo, + RequestId, + Transport +} from '@modelcontextprotocol/core-internal'; import { DEFAULT_NEGOTIATED_PROTOCOL_VERSION, isInitializeRequest, isJsonContentType, isJSONRPCErrorResponse, + isJSONRPCNotification, isJSONRPCRequest, isJSONRPCResultResponse, JSONRPCMessageSchema, @@ -910,7 +919,7 @@ export class WebStandardStreamableHTTPServerTransport implements Transport { if (!hasRequests) { // if it only contains notifications or responses, return 202 for (const message of messages) { - this.onmessage?.(message, { authInfo: options?.authInfo, request }); + this._deliverMessage(message, { authInfo: options?.authInfo, request }); } return new Response(null, { status: 202 }); } @@ -944,7 +953,7 @@ export class WebStandardStreamableHTTPServerTransport implements Transport { } for (const message of messages) { - this.onmessage?.(message, { authInfo: options?.authInfo, request }); + this._deliverMessage(message, { authInfo: options?.authInfo, request }); } }); } @@ -1024,7 +1033,7 @@ export class WebStandardStreamableHTTPServerTransport implements Transport { }; } - this.onmessage?.(message, { authInfo: options?.authInfo, request, closeSSEStream, closeStandaloneSSEStream }); + this._deliverMessage(message, { authInfo: options?.authInfo, request, closeSSEStream, closeStandaloneSSEStream }); } // 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 @@ -1040,6 +1049,70 @@ export class WebStandardStreamableHTTPServerTransport implements Transport { } } + /** + * Dispatches a message, then retires the mapped stream of a cancelled + * request. A cancellation replaces that request's response, so nothing + * else will settle its POST or close its SSE stream. + */ + private _deliverMessage(message: JSONRPCMessage, extra: MessageExtraInfo): void { + this.onmessage?.(message, extra); + + if (isJSONRPCNotification(message) && message.method === 'notifications/cancelled') { + this._settleCancelledRequest((message.params as CancelledNotificationParams | undefined)?.requestId); + } + } + + /** + * Drops a cancelled id from the request bookkeeping. When no other + * in-flight request needs the same stream, the stream is completed with an + * empty 202 in JSON mode, or closed in SSE mode. Batched siblings that are + * still pending keep the stream open. + */ + private _settleCancelledRequest(requestId: RequestId | undefined): void { + if (requestId === undefined) { + return; + } + + const streamId = this._requestToStreamMapping.get(requestId); + if (streamId === undefined) { + return; + } + + this._requestToStreamMapping.delete(requestId); + this._requestResponseMap.delete(requestId); + + const relatedIds = [...this._requestToStreamMapping.entries()].filter(([, sid]) => sid === streamId).map(([id]) => id); + if (!relatedIds.every(id => this._requestResponseMap.has(id))) { + return; + } + + const stream = this._streamMapping.get(streamId); + if (stream === undefined) { + return; + } + + if (this._enableJsonResponse && stream.resolveJson !== undefined) { + const headers: Record = {}; + if (this.sessionId !== undefined) { + headers['mcp-session-id'] = this.sessionId; + } + + if (relatedIds.length === 0) { + stream.resolveJson(new Response(null, { status: 202, headers })); + } else { + headers['Content-Type'] = 'application/json'; + const responses = relatedIds.map(id => this._requestResponseMap.get(id)!); + stream.resolveJson(Response.json(responses.length === 1 ? responses[0] : responses, { status: 200, headers })); + } + } + + stream.cleanup(); + for (const id of relatedIds) { + this._requestResponseMap.delete(id); + this._requestToStreamMapping.delete(id); + } + } + /** * Handles `DELETE` requests to terminate sessions */ diff --git a/packages/server/test/server/streamableHttp.test.ts b/packages/server/test/server/streamableHttp.test.ts index 937316ebed..b5a7c091a8 100644 --- a/packages/server/test/server/streamableHttp.test.ts +++ b/packages/server/test/server/streamableHttp.test.ts @@ -574,6 +574,123 @@ describe('Zod v4', () => { }); }); + describe('HTTPServerTransport - Cancellation', () => { + async function createTransport(enableJsonResponse: boolean): Promise { + const mcpServer = new McpServer({ name: 'test-server', version: '1.0.0' }); + + // Resolves only when the request is cancelled, so the POST can never + // settle by the handler finishing. + mcpServer.registerTool( + 'hang', + { description: 'Waits for its request to be cancelled', inputSchema: z.object({}) }, + async (_args, ctx) => { + await new Promise(resolve => { + ctx.mcpReq.signal.addEventListener('abort', () => resolve(), { once: true }); + }); + return { content: [] }; + } + ); + mcpServer.registerTool('fast', { description: 'Returns immediately' }, async () => ({ content: [] })); + + const transport = new WebStandardStreamableHTTPServerTransport({ + sessionIdGenerator: () => randomUUID(), + enableJsonResponse + }); + await mcpServer.connect(transport); + return transport; + } + + async function initialize(transport: WebStandardStreamableHTTPServerTransport): Promise { + const response = await transport.handleRequest(createRequest('POST', TEST_MESSAGES.initialize)); + expect(response.status).toBe(200); + const sessionId = response.headers.get('mcp-session-id'); + expect(sessionId).toBeDefined(); + return sessionId as string; + } + + function cancelRequest(sessionId: string, requestId: string): Request { + return createRequest('POST', { jsonrpc: '2.0', method: 'notifications/cancelled', params: { requestId } }, { sessionId }); + } + + it('settles a cancelled JSON-mode POST with an empty 202 and removes its mappings', async () => { + const transport = await createTransport(true); + try { + const sessionId = await initialize(transport); + + const call = transport.handleRequest( + createRequest( + 'POST', + { jsonrpc: '2.0', method: 'tools/call', params: { name: 'hang' }, id: 'cancel-me' }, + { sessionId } + ) + ); + // Let the POST register its stream and reach the handler. + await new Promise(resolve => setImmediate(resolve)); + + const cancel = await transport.handleRequest(cancelRequest(sessionId, 'cancel-me')); + expect(cancel.status).toBe(202); + + const response = await call; + expect(response.status).toBe(202); + expect(await response.text()).toBe(''); + expect((transport as any)._streamMapping.size).toBe(0); + expect((transport as any)._requestToStreamMapping.size).toBe(0); + } finally { + await transport.close(); + } + }); + + it('closes a cancelled per-request SSE stream and removes its mappings', async () => { + const transport = await createTransport(false); + try { + const sessionId = await initialize(transport); + + const response = await transport.handleRequest( + createRequest( + 'POST', + { jsonrpc: '2.0', method: 'tools/call', params: { name: 'hang' }, id: 'cancel-me' }, + { sessionId } + ) + ); + expect(response.status).toBe(200); + const reader = response.body!.getReader(); + await new Promise(resolve => setImmediate(resolve)); + + await transport.handleRequest(cancelRequest(sessionId, 'cancel-me')); + + expect((await reader.read()).done).toBe(true); + expect((transport as any)._streamMapping.size).toBe(0); + expect((transport as any)._requestToStreamMapping.size).toBe(0); + } finally { + await transport.close(); + } + }); + + it('still delivers a completed batch sibling when the cancelled id is retired', async () => { + const transport = await createTransport(true); + try { + const sessionId = await initialize(transport); + + const batch: JSONRPCMessage[] = [ + { jsonrpc: '2.0', method: 'tools/call', params: { name: 'hang' }, id: 'cancel-me' }, + { jsonrpc: '2.0', method: 'tools/call', params: { name: 'fast' }, id: 'fast-1' } + ]; + const pending = transport.handleRequest(createRequest('POST', batch, { sessionId })); + await new Promise(resolve => setImmediate(resolve)); + + await transport.handleRequest(cancelRequest(sessionId, 'cancel-me')); + + const response = await pending; + expect(response.status).toBe(200); + expect(await response.json()).toMatchObject({ jsonrpc: '2.0', id: 'fast-1' }); + expect((transport as any)._streamMapping.size).toBe(0); + expect((transport as any)._requestToStreamMapping.size).toBe(0); + } finally { + await transport.close(); + } + }); + }); + describe('HTTPServerTransport - Session Callbacks', () => { it('should call onsessioninitialized callback', async () => { const onInitialized = vi.fn(); From d9d5d821dea2c276a68a4e29ecf20b628391c87a Mon Sep 17 00:00:00 2001 From: Sanath Vengaldas Date: Mon, 26 Oct 2026 13:16:07 +0000 Subject: [PATCH 2/3] chore: add changeset --- .changeset/cancelled-post-settles.md | 5 +++++ 1 file changed, 5 insertions(+) create mode 100644 .changeset/cancelled-post-settles.md diff --git a/.changeset/cancelled-post-settles.md b/.changeset/cancelled-post-settles.md new file mode 100644 index 0000000000..3f28a04f10 --- /dev/null +++ b/.changeset/cancelled-post-settles.md @@ -0,0 +1,5 @@ +--- +'@modelcontextprotocol/server': patch +--- + +Settle a cancelled request's POST and release its stream mapping, instead of leaving the response pending forever. From b7f7c2a2046e14e6bd79710138c242811d8889e6 Mon Sep 17 00:00:00 2001 From: Sanath Vengaldas Date: Thu, 29 Oct 2026 04:49:33 +0000 Subject: [PATCH 3/3] fix(server): retire answered siblings when the cancelled stream is gone _settleCancelledRequest returned before the bookkeeping cleanup when the per-request SSE stream had already been closed or dropped. In a batch where an answered sibling was parked in _requestResponseMap waiting on the cancelled id, that left both of its mappings behind - the same leak this PR fixes, on the disconnected path. send() already retires relatedIds when the stream is missing; do the same here and skip only the resolveJson/cleanup part. Adds a test that closes the stream from the first tool, cancels the second, and asserts both maps are empty and the answered response is still in the event store. --- packages/server/src/server/streamableHttp.ts | 33 +++++----- .../server/test/server/streamableHttp.test.ts | 65 +++++++++++++++++++ 2 files changed, 83 insertions(+), 15 deletions(-) diff --git a/packages/server/src/server/streamableHttp.ts b/packages/server/src/server/streamableHttp.ts index 8bebc53b17..16c985e967 100644 --- a/packages/server/src/server/streamableHttp.ts +++ b/packages/server/src/server/streamableHttp.ts @@ -1087,26 +1087,29 @@ export class WebStandardStreamableHTTPServerTransport implements Transport { } const stream = this._streamMapping.get(streamId); - if (stream === undefined) { - return; - } + if (stream !== undefined) { + if (this._enableJsonResponse && stream.resolveJson !== undefined) { + const headers: Record = {}; + if (this.sessionId !== undefined) { + headers['mcp-session-id'] = this.sessionId; + } - if (this._enableJsonResponse && stream.resolveJson !== undefined) { - const headers: Record = {}; - if (this.sessionId !== undefined) { - headers['mcp-session-id'] = this.sessionId; + if (relatedIds.length === 0) { + stream.resolveJson(new Response(null, { status: 202, headers })); + } else { + headers['Content-Type'] = 'application/json'; + const responses = relatedIds.map(id => this._requestResponseMap.get(id)!); + stream.resolveJson(Response.json(responses.length === 1 ? responses[0] : responses, { status: 200, headers })); + } } - if (relatedIds.length === 0) { - stream.resolveJson(new Response(null, { status: 202, headers })); - } else { - headers['Content-Type'] = 'application/json'; - const responses = relatedIds.map(id => this._requestResponseMap.get(id)!); - stream.resolveJson(Response.json(responses.length === 1 ? responses[0] : responses, { status: 200, headers })); - } + stream.cleanup(); } - stream.cleanup(); + // Retire answered siblings even when there is no live stream: with an + // eventStore their responses were stored for replay, and without one + // nothing is left that could deliver them — keeping the ids would only + // leak. for (const id of relatedIds) { this._requestResponseMap.delete(id); this._requestToStreamMapping.delete(id); diff --git a/packages/server/test/server/streamableHttp.test.ts b/packages/server/test/server/streamableHttp.test.ts index b5a7c091a8..39361bf228 100644 --- a/packages/server/test/server/streamableHttp.test.ts +++ b/packages/server/test/server/streamableHttp.test.ts @@ -942,6 +942,71 @@ describe('Zod v4', () => { expect((transport as any)._requestToStreamMapping.has('disconnect-1')).toBe(false); }); + it('retires an answered batch sibling when the cancelled id has no live stream left', async () => { + // Batch [answered, cancelled]: the answered sibling's response is + // stored and parked in _requestResponseMap while the cancelled id + // is still pending. The handler closed the per-request stream, so + // by the time the cancellation settles the batch there is no + // _streamMapping entry. The answered id must still be retired — + // send() does the same when the stream is missing. + mcpServer.registerTool( + 'closeThenReturn', + { description: 'closeSSE then return', inputSchema: z.object({}) }, + async (_args, ctx): Promise => { + ctx.http?.closeSSE?.(); + return { content: [{ type: 'text', text: 'answered' }] }; + } + ); + mcpServer.registerTool( + 'hangUntilCancelled', + { description: 'wait for cancellation', inputSchema: z.object({}) }, + async (_args, ctx): Promise => { + await new Promise(resolve => { + ctx.mcpReq.signal.addEventListener('abort', () => resolve(), { once: true }); + }); + return { content: [] }; + } + ); + + sessionId = await initializeServer(); + storedEvents.clear(); + + const batch: JSONRPCMessage[] = [ + { jsonrpc: '2.0', method: 'tools/call', params: { name: 'closeThenReturn', arguments: {} }, id: 'answered' }, + { jsonrpc: '2.0', method: 'tools/call', params: { name: 'hangUntilCancelled', arguments: {} }, id: 'cancel-me' } + ]; + const response = await transport.handleRequest(createRequest('POST', batch, { sessionId })); + await response.text().catch(() => {}); + + // Wait until the answered sibling is parked and its stream is gone. + for (let i = 0; i < 100; i++) { + // eslint-disable-next-line @typescript-eslint/no-explicit-any + if ((transport as any)._requestResponseMap.has('answered')) break; + await new Promise(resolve => setTimeout(resolve, 5)); + } + // eslint-disable-next-line @typescript-eslint/no-explicit-any + expect((transport as any)._streamMapping.size).toBe(0); + // eslint-disable-next-line @typescript-eslint/no-explicit-any + expect((transport as any)._requestResponseMap.has('answered')).toBe(true); + + const cancel = await transport.handleRequest( + createRequest( + 'POST', + { jsonrpc: '2.0', method: 'notifications/cancelled', params: { requestId: 'cancel-me' } }, + { sessionId } + ) + ); + expect(cancel.status).toBe(202); + + // eslint-disable-next-line @typescript-eslint/no-explicit-any + expect((transport as any)._requestResponseMap.size).toBe(0); + // eslint-disable-next-line @typescript-eslint/no-explicit-any + expect((transport as any)._requestToStreamMapping.size).toBe(0); + + const stored = [...storedEvents.values()].map(e => e.message); + expect(stored.some(m => 'id' in m && m.id === 'answered' && 'result' in m)).toBe(true); + }); + it('should accept Last-Event-ID reconnect after closeSSEStream() and replay stored events', async () => { // closeSSEStream() removes the _streamMapping entry, so the // replayEvents() conflict check sees no active connection and the