diff --git a/.changeset/answer-in-flight-requests-on-transport-close.md b/.changeset/answer-in-flight-requests-on-transport-close.md new file mode 100644 index 0000000000..719bee7478 --- /dev/null +++ b/.changeset/answer-in-flight-requests-on-transport-close.md @@ -0,0 +1,31 @@ +--- +'@modelcontextprotocol/server': patch +--- + +Settle pending JSON-response-mode requests when the streamable HTTP transport +closes, instead of leaving the HTTP request hanging. + +In JSON response mode, `handleRequest()` returns a `Promise` that only +`send()` resolves. A stream mapping's `cleanup` deletes the entry without +settling that promise, so `WebStandardStreamableHTTPServerTransport.close()` +while a POST was in flight left the HTTP request open until the socket died — +and the caller unable to tell whether the request ran, which for a mutating call +means retrying may double-execute and giving up may drop a completed write. + +`close()` now resolves any such pending response before the stream mappings are +torn down, with a JSON-RPC error (`-32000`, `"Connection closed"`) for each +request id still outstanding. A batched POST shares one stream across several +request ids, so the outstanding ids are grouped by stream, and an id that +already has a real response reuses it rather than being overwritten with an +error. + +The SSE path is deliberately unchanged: a POST-initiated SSE stream closing +without a JSON-RPC response is the documented outcome of session termination +(`hosting:session:delete-cancels-inflight`), where the request handler has been +aborted and the request is therefore cancelled rather than unanswered. + +`close()` also clears `_requestToStreamMapping`, which it previously left +populated. + +`NodeStreamableHTTPServerTransport` wraps this transport, so it inherits the +fix. diff --git a/packages/server/src/server/streamableHttp.ts b/packages/server/src/server/streamableHttp.ts index c0f48560a2..4f22741bfc 100644 --- a/packages/server/src/server/streamableHttp.ts +++ b/packages/server/src/server/streamableHttp.ts @@ -1048,12 +1048,87 @@ export class WebStandardStreamableHTTPServerTransport implements Transport { return undefined; } + /** + * Settles the pending `Promise` of any JSON-response-mode POST + * that is still in flight, so closing the transport cannot leave an HTTP + * request hanging. + * + * In JSON response mode `handleRequest()` returns a promise that only + * `send()` resolves. A stream mapping's `cleanup` deletes the entry + * without settling that promise, so a `close()` while a POST is in flight + * left the HTTP request open until the socket died, with the caller unable + * to tell whether the request ran. + * + * The SSE path is deliberately not touched: a POST-initiated SSE stream + * closing without a JSON-RPC response is the documented outcome of session + * termination (`hosting:session:delete-cancels-inflight`), where the + * request handler has been aborted and the request is therefore cancelled + * rather than unanswered. + */ + private settlePendingJsonResponses(): void { + if (!this._enableJsonResponse || this._requestToStreamMapping.size === 0) { + return; + } + + // Group the outstanding ids by the stream that owes them a response: + // a batched POST shares one stream across several request ids. + const idsByStream = new Map(); + for (const [requestId, streamId] of this._requestToStreamMapping) { + const ids = idsByStream.get(streamId); + if (ids === undefined) { + idsByStream.set(streamId, [requestId]); + } else { + ids.push(requestId); + } + } + + for (const [streamId, requestIds] of idsByStream) { + const stream = this._streamMapping.get(streamId); + if (stream?.resolveJson === undefined) { + continue; + } + + // An id that already has a response only reaches here as part of a + // batch whose siblings are unanswered — reuse the real response + // rather than overwriting it with an error. + const messages = requestIds.map( + requestId => + this._requestResponseMap.get(requestId) ?? + ({ + jsonrpc: '2.0', + id: requestId, + error: { + code: -32_000, + message: 'Connection closed: the server transport closed before this request completed' + } + } satisfies JSONRPCMessage) + ); + + const headers: Record = { 'Content-Type': 'application/json' }; + if (this.sessionId !== undefined) { + headers['mcp-session-id'] = this.sessionId; + } + + try { + stream.resolveJson(Response.json(messages.length === 1 ? messages[0] : messages, { status: 200, headers })); + } catch (error) { + // Never let one undeliverable response stop the others being + // settled, or close() from completing. + this.onerror?.(error as Error); + } + } + } + async close(): Promise { if (this._closed) { return; } this._closed = true; + // Settle pending JSON-mode responses BEFORE the streams are torn + // down — cleanup() drops the mapping without resolving the promise. + this.settlePendingJsonResponses(); + // Close all SSE connections for (const { cleanup } of this._streamMapping.values()) { cleanup(); @@ -1061,6 +1136,7 @@ export class WebStandardStreamableHTTPServerTransport implements Transport { this._streamMapping.clear(); // Clear any pending responses + this._requestToStreamMapping.clear(); this._requestResponseMap.clear(); this.onclose?.(); } diff --git a/packages/server/test/server/streamableHttp.test.ts b/packages/server/test/server/streamableHttp.test.ts index 9ec6baf46c..29512839c2 100644 --- a/packages/server/test/server/streamableHttp.test.ts +++ b/packages/server/test/server/streamableHttp.test.ts @@ -1370,6 +1370,70 @@ describe('Zod v4', () => { }); }); + describe('close() with a JSON-mode request in flight', () => { + // In JSON response mode handleRequest() returns a promise that only + // send() resolves; cleanup() drops the mapping without settling it, so + // close() during an in-flight POST left the HTTP request hanging. + async function createUnansweringTransport(): Promise { + const transport = new WebStandardStreamableHTTPServerTransport({ + sessionIdGenerator: () => randomUUID(), + enableJsonResponse: true + }); + // No server attached: every request stays in flight forever. + transport.onmessage = () => {}; + await transport.start(); + return transport; + } + + it('resolves the pending response with a JSON-RPC error', async () => { + const transport = await createUnansweringTransport(); + + const pending = transport.handleRequest(createRequest('POST', TEST_MESSAGES.initialize)); + // Give handleRequest a turn to register the stream before closing. + await new Promise(resolve => setImmediate(resolve)); + + await transport.close(); + + const response = await pending; + expect(response.status).toBe(200); + const data = (await response.json()) as JSONRPCErrorResponse; + expect(data.id).toBe('init-1'); + expectErrorResponse(data, -32_000, /Connection closed/); + }); + + it('resolves every id of an in-flight batch', async () => { + // Stateless: no initialize handshake to await, which in JSON mode + // would itself never resolve without a server attached. + const transport = new WebStandardStreamableHTTPServerTransport({ + sessionIdGenerator: undefined, + enableJsonResponse: true + }); + transport.onmessage = () => {}; + await transport.start(); + + const batch: JSONRPCMessage[] = [ + { jsonrpc: '2.0', method: 'tools/list', params: {}, id: 'batch-1' }, + { jsonrpc: '2.0', method: 'tools/list', params: {}, id: 'batch-2' } + ]; + const pending = transport.handleRequest(createRequest('POST', batch)); + await new Promise(resolve => setImmediate(resolve)); + + await transport.close(); + + const data = (await (await pending).json()) as JSONRPCErrorResponse[]; + expect(data.map(entry => entry.id)).toEqual(['batch-1', 'batch-2']); + }); + + it('is a no-op when nothing is in flight', async () => { + const transport = await createUnansweringTransport(); + const onerror = vi.fn(); + transport.onerror = onerror; + + await expect(transport.close()).resolves.toBeUndefined(); + expect(onerror).not.toHaveBeenCalled(); + }); + }); + describe('close() re-entrancy guard', () => { it('should not recurse when onclose triggers a second close()', async () => { const transport = new WebStandardStreamableHTTPServerTransport({ sessionIdGenerator: randomUUID });