From 21cd55255b1bd5fffcc8b74a001c52ce69b15afc Mon Sep 17 00:00:00 2001 From: Sammy Dabbas Date: Thu, 9 Jul 2026 12:07:32 -0400 Subject: [PATCH] fix(server): reject duplicate in-flight request ids in streamable HTTP Backport of #2434 to the v1.x web-standard transport. The transport registered every request via _requestToStreamMapping.set(message.id, streamId) with no duplicate check, so a concurrent POST reusing an in-flight id overwrote the entry, cross-wired the first request's response onto the second POST's stream, and left the first POST hanging. Reject a POST containing a request whose id is already in flight, or duplicated within the same batch, with HTTP 400 and JSON-RPC -32600. Sequential reuse after completion stays allowed since deployed clients send a constant id for every request. Cancelled requests never produce a response, so notifications/cancelled now retires the transport bookkeeping for its target id; without that, a cancelled id would stay in flight forever and lock out cancel-then-reuse clients. Adds the missing isCancelledNotification guard to types. Fixes #2433. --- .changeset/fix-duplicate-inflight-ids.md | 7 + src/server/webStandardStreamableHttp.ts | 33 ++++ src/types.ts | 3 + test/server/streamableHttp.test.ts | 211 +++++++++++++++++++++++ 4 files changed, 254 insertions(+) create mode 100644 .changeset/fix-duplicate-inflight-ids.md diff --git a/.changeset/fix-duplicate-inflight-ids.md b/.changeset/fix-duplicate-inflight-ids.md new file mode 100644 index 0000000000..63855ef233 --- /dev/null +++ b/.changeset/fix-duplicate-inflight-ids.md @@ -0,0 +1,7 @@ +--- +'@modelcontextprotocol/sdk': patch +--- + +Reject duplicate in-flight request ids in the Streamable HTTP server transport instead of cross-wiring responses. A second POST that reused a JSON-RPC id still in flight previously overwrote the `_requestToStreamMapping` entry, routing the original request's response onto the new +stream and leaving the first POST hanging; such requests, and ids duplicated within a single batch, are now rejected with HTTP 400 and JSON-RPC -32600. Transport bookkeeping for cancelled requests is retired on `notifications/cancelled` so their ids stay reusable, and a +`isCancelledNotification` guard is added. diff --git a/src/server/webStandardStreamableHttp.ts b/src/server/webStandardStreamableHttp.ts index 5721d3e370..ad78f72402 100644 --- a/src/server/webStandardStreamableHttp.ts +++ b/src/server/webStandardStreamableHttp.ts @@ -13,6 +13,7 @@ import { AuthInfo } from './auth/types.js'; import { MessageExtraInfo, RequestInfo, + isCancelledNotification, isInitializeRequest, isJSONRPCErrorResponse, isJSONRPCRequest, @@ -688,6 +689,38 @@ export class WebStandardStreamableHTTPServerTransport implements Transport { } } + // A cancelled request never produces a response (the protocol layer + // suppresses responses for aborted handlers), so retire its transport + // bookkeeping here. Otherwise its _requestToStreamMapping entry would + // pin the request ID as in-flight forever and the duplicate-ID guard + // below would reject any later reuse of that ID. + for (const message of messages) { + if (isCancelledNotification(message) && message.params.requestId !== undefined) { + this._requestToStreamMapping.delete(message.params.requestId); + this._requestResponseMap.delete(message.params.requestId); + } + } + + // Request IDs MUST be unique within a session. Registering a duplicate + // would overwrite the in-flight entry in _requestToStreamMapping and + // cross-wire the original request's response onto this POST's stream, + // leaving the original POST hanging - so reject the duplicate while the + // first request is still in flight. Sequential reuse stays allowed: + // entries are retired when the response is delivered or the request is + // cancelled. + const incomingRequestIds = new Set(); + for (const message of messages) { + if (!isJSONRPCRequest(message)) { + continue; + } + if (this._requestToStreamMapping.has(message.id) || incomingRequestIds.has(message.id)) { + const error = `Invalid Request: Request ID ${String(message.id)} is already in flight`; + this.onerror?.(new Error(error)); + return this.createJsonErrorResponse(400, -32600, error); + } + incomingRequestIds.add(message.id); + } + // check if it contains requests const hasRequests = messages.some(isJSONRPCRequest); diff --git a/src/types.ts b/src/types.ts index 835eac89f8..0f9e8c3757 100644 --- a/src/types.ts +++ b/src/types.ts @@ -629,6 +629,9 @@ export const InitializedNotificationSchema = NotificationSchema.extend({ export const isInitializedNotification = (value: unknown): value is InitializedNotification => InitializedNotificationSchema.safeParse(value).success; +export const isCancelledNotification = (value: unknown): value is CancelledNotification => + CancelledNotificationSchema.safeParse(value).success; + /* Ping */ /** * A ping, issued by either the server or the client, to check that the other party is still alive. The receiver must promptly respond, or else may be disconnected. diff --git a/test/server/streamableHttp.test.ts b/test/server/streamableHttp.test.ts index 4a4f7d8248..d02dcd8609 100644 --- a/test/server/streamableHttp.test.ts +++ b/test/server/streamableHttp.test.ts @@ -3272,3 +3272,214 @@ describe('WebStandardStreamableHTTPServerTransport - onerror callback', () => { await storeTransport.close(); }); }); + +describe('WebStandardStreamableHTTPServerTransport - Duplicate request IDs', () => { + let transport: WebStandardStreamableHTTPServerTransport; + let mcpServer: McpServer; + let slowToolStarted: Promise; + let releaseSlowTool: () => void; + + /** 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 ? (typeof opts.body === 'string' ? opts.body : JSON.stringify(opts.body)) : undefined + }); + } + + function withSession(sessionId: string, extra?: Record): Record { + return { 'mcp-session-id': sessionId, 'mcp-protocol-version': '2025-11-25', ...extra }; + } + + /** Extract the JSON-RPC payload from a single SSE `data:` line. */ + function parseSSEData(raw: string): unknown { + const dataLine = raw.split('\n').find(line => line.startsWith('data:')); + if (!dataLine) { + throw new Error(`No SSE data line found in: ${raw}`); + } + return JSON.parse(dataLine.slice('data:'.length).trim()); + } + + function callTool(name: string, id: string): JSONRPCMessage { + return { + jsonrpc: '2.0', + method: 'tools/call', + params: { name, arguments: {} }, + id + } as JSONRPCMessage; + } + + function setupServer(options?: { enableJsonResponse?: boolean }): void { + mcpServer = new McpServer({ name: 'test-server', version: '1.0.0' }, { capabilities: { logging: {} } }); + + mcpServer.tool( + 'greet', + 'Greeting tool', + async (): Promise => ({ content: [{ type: 'text', text: 'Hello, Test!' }] }) + ); + + let started!: () => void; + slowToolStarted = new Promise(resolve => { + started = resolve; + }); + const gate = new Promise(resolve => { + releaseSlowTool = resolve; + }); + mcpServer.tool('slow', 'A tool that blocks until released by the test', async (): Promise => { + started(); + await gate; + return { content: [{ type: 'text', text: 'slow-done' }] }; + }); + + transport = new WebStandardStreamableHTTPServerTransport({ + sessionIdGenerator: () => randomUUID(), + enableJsonResponse: options?.enableJsonResponse ?? false + }); + } + + beforeEach(async () => { + setupServer(); + await mcpServer.connect(transport); + }); + + afterEach(async () => { + releaseSlowTool(); + await transport.close(); + }); + + async function initializeServer(): Promise { + const response = await transport.handleRequest(req('POST', { body: TEST_MESSAGES.initialize })); + expect(response.status).toBe(200); + const newSessionId = response.headers.get('mcp-session-id'); + expect(newSessionId).toBeDefined(); + return newSessionId as string; + } + + it('should reject a request whose ID collides with one still in flight', async () => { + const sessionId = await initializeServer(); + + // First request occupies the ID and blocks inside the tool handler + const firstResponse = await transport.handleRequest( + req('POST', { body: callTool('slow', 'dup-1'), headers: withSession(sessionId) }) + ); + expect(firstResponse.status).toBe(200); + await slowToolStarted; + + // Concurrent request reusing the same ID must be rejected, not cross-wired + const secondResponse = await transport.handleRequest( + req('POST', { body: callTool('greet', 'dup-1'), headers: withSession(sessionId) }) + ); + expect(secondResponse.status).toBe(400); + const errorData = await secondResponse.json(); + expectErrorResponse(errorData, -32600, /already in flight/); + + // The original request must still receive its own response + releaseSlowTool(); + const eventData = parseSSEData(await readSSEEvent(firstResponse)); + expect(eventData).toMatchObject({ + jsonrpc: '2.0', + result: { content: [{ type: 'text', text: 'slow-done' }] }, + id: 'dup-1' + }); + }); + + it('should reject duplicate request IDs within a single batch', async () => { + const sessionId = await initializeServer(); + + const batch: JSONRPCMessage[] = [callTool('greet', 'batch-dup'), callTool('greet', 'batch-dup')]; + const response = await transport.handleRequest(req('POST', { body: batch, headers: withSession(sessionId) })); + + expect(response.status).toBe(400); + const errorData = await response.json(); + expectErrorResponse(errorData, -32600, /already in flight/); + }); + + it('should allow sequential reuse of a request ID after the response is delivered', async () => { + const sessionId = await initializeServer(); + + for (let attempt = 0; attempt < 2; attempt++) { + const response = await transport.handleRequest( + req('POST', { body: callTool('greet', 'reuse-1'), headers: withSession(sessionId) }) + ); + expect(response.status).toBe(200); + + const eventData = parseSSEData(await readSSEEvent(response)); + expect(eventData).toMatchObject({ + jsonrpc: '2.0', + result: { content: [{ type: 'text', text: 'Hello, Test!' }] }, + id: 'reuse-1' + }); + } + }); + + it('should allow reuse of a request ID after the original request was cancelled', async () => { + const sessionId = await initializeServer(); + + const firstResponse = await transport.handleRequest( + req('POST', { body: callTool('slow', 'cancel-1'), headers: withSession(sessionId) }) + ); + expect(firstResponse.status).toBe(200); + await slowToolStarted; + + const cancelNotification: JSONRPCMessage = { + jsonrpc: '2.0', + method: 'notifications/cancelled', + params: { requestId: 'cancel-1', reason: 'test cancellation' } + }; + const cancelResponse = await transport.handleRequest(req('POST', { body: cancelNotification, headers: withSession(sessionId) })); + expect(cancelResponse.status).toBe(202); + + // The cancelled request will never produce a response, so its ID must be reusable + const secondResponse = await transport.handleRequest( + req('POST', { body: callTool('greet', 'cancel-1'), headers: withSession(sessionId) }) + ); + expect(secondResponse.status).toBe(200); + + const eventData = parseSSEData(await readSSEEvent(secondResponse)); + expect(eventData).toMatchObject({ + jsonrpc: '2.0', + result: { content: [{ type: 'text', text: 'Hello, Test!' }] }, + id: 'cancel-1' + }); + }); + + it('should reject duplicate in-flight request IDs in JSON response mode', async () => { + await transport.close(); + setupServer({ enableJsonResponse: true }); + await mcpServer.connect(transport); + + const sessionId = await initializeServer(); + + // In JSON mode handleRequest resolves only when the response is ready - do not await + const firstResponsePromise = transport.handleRequest( + req('POST', { body: callTool('slow', 'dup-json'), headers: withSession(sessionId) }) + ); + await slowToolStarted; + + const secondResponse = await transport.handleRequest( + req('POST', { body: callTool('greet', 'dup-json'), headers: withSession(sessionId) }) + ); + expect(secondResponse.status).toBe(400); + const errorData = await secondResponse.json(); + expectErrorResponse(errorData, -32600, /already in flight/); + + // The original request must still resolve with its own result + releaseSlowTool(); + const firstResponse = await firstResponsePromise; + expect(firstResponse.status).toBe(200); + const data = await firstResponse.json(); + expect(data).toMatchObject({ + jsonrpc: '2.0', + result: { content: [{ type: 'text', text: 'slow-done' }] }, + id: 'dup-json' + }); + }); +});