fix(node): cancel the response stream when the client disconnects

.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.

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<void>(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) {

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