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. diff --git a/packages/server/src/server/streamableHttp.ts b/packages/server/src/server/streamableHttp.ts index 2484ae19b0..16c985e967 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,73 @@ 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) { + 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(); + } + + // 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); + } + } + /** * 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..39361bf228 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(); @@ -825,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