From 6ca7b5b5f54dc7000da308fbfcaa7501a6a32f9c Mon Sep 17 00:00:00 2001 From: SwapnilSkr Date: Thu, 5 Nov 2026 11:22:57 +0530 Subject: [PATCH] fix(node): cancel the response stream when the client disconnects `toNodeHandler` read the response body with `for await`, so it only saw a client disconnect when the stream produced its next chunk. On an idle GET stream that held the session's only GET slot until the next keep-alive, and reconnects got 409 until the client gave up. Read through a reader and cancel it on abort. Racing `next()` against the abort does not work here: `return()` on the async iterator waits for the pending `next()`, so an idle stream would still not be cancelled. --- .../node-handler-cancel-on-disconnect.md | 5 +++ packages/middleware/node/src/toNodeHandler.ts | 16 ++++++--- .../node/test/toNodeHandler.test.ts | 34 +++++++++++++++++++ 3 files changed, 51 insertions(+), 4 deletions(-) create mode 100644 .changeset/node-handler-cancel-on-disconnect.md diff --git a/.changeset/node-handler-cancel-on-disconnect.md b/.changeset/node-handler-cancel-on-disconnect.md new file mode 100644 index 0000000000..725ef731a5 --- /dev/null +++ b/.changeset/node-handler-cancel-on-disconnect.md @@ -0,0 +1,5 @@ +--- +'@modelcontextprotocol/node': patch +--- + +`toNodeHandler` cancels the response stream as soon as the client disconnects, instead of at the stream's next chunk. A session's GET stream was held until the next keep-alive write, so a client reconnecting after a dropped connection got 409 and could give up. diff --git a/packages/middleware/node/src/toNodeHandler.ts b/packages/middleware/node/src/toNodeHandler.ts index 5f26b1ad26..a5561aab5b 100644 --- a/packages/middleware/node/src/toNodeHandler.ts +++ b/packages/middleware/node/src/toNodeHandler.ts @@ -188,8 +188,8 @@ export function toNodeHandler(handler: FetchLikeMcpHandler, opts?: ToNodeHandler // from `res.destroyed` at entry to cover the pre-registration window // when 'close' already fired during async middleware) is the single // termination source — racing it against the drain wait means a - // vanished client cannot park the loop, and breaking out of the async - // iterator calls return() to cancel the upstream stream. + // vanished client cannot park the loop, and cancelling the reader + // releases an idle upstream stream without waiting for its next chunk. let drainResolve: (() => void) | undefined; const releaseDrainWait = () => { drainResolve?.(); @@ -199,9 +199,17 @@ export function toNodeHandler(handler: FetchLikeMcpHandler, opts?: ToNodeHandler const closed = new Promise(resolve => { abort.signal.addEventListener('abort', () => resolve(), { once: true }); }); + const reader = response.body.getReader(); + const cancelBody = () => void reader.cancel().catch(() => {}); + if (abort.signal.aborted) { + cancelBody(); + } else { + abort.signal.addEventListener('abort', cancelBody, { once: true }); + } try { - for await (const chunk of response.body) { - if (abort.signal.aborted) { + while (true) { + const { done, value: chunk } = await reader.read(); + if (done || abort.signal.aborted) { break; } if (res.write(chunk) === false) { diff --git a/packages/middleware/node/test/toNodeHandler.test.ts b/packages/middleware/node/test/toNodeHandler.test.ts index b879365f4a..26d1d594f3 100644 --- a/packages/middleware/node/test/toNodeHandler.test.ts +++ b/packages/middleware/node/test/toNodeHandler.test.ts @@ -242,6 +242,40 @@ describe('toNodeHandler', () => { ).resolves.toBeUndefined(); }); + it('cancels an idle response stream as soon as the client closes', async () => { + const cancel = vi.fn(); + const node = toNodeHandler({ fetch: async () => new Response(new ReadableStream({ cancel })) }); + + const closeListeners: Array<(...args: unknown[]) => void> = []; + let headersWritten = false; + const res: NodeServerResponseLike = { + writeHead() { + headersWritten = true; + return this; + }, + write() { + return true; + }, + end() { + return this; + }, + on(event: string, listener: (...args: unknown[]) => void) { + if (event === 'close') closeListeners.push(listener); + return this; + } + }; + + const { req } = nodeRequestResponse({}); + const served = node(req, res); + await vi.waitFor(() => expect(headersWritten).toBe(true)); + + // No chunk ever arrives, as on a session GET stream between keep-alives. + for (const listener of closeListeners) listener(); + + await served; + expect(cancel).toHaveBeenCalledOnce(); + }); + it('does not park on backpressure when the response was already destroyed before the adapter listened', async () => { // 'close' fired during async middleware BEFORE toNodeHandler registered // its listener — `res.destroyed` is the entry-time witness. The