diff --git a/docs/specs/agent-environments/http-transport/verification.md b/docs/specs/agent-environments/http-transport/verification.md index 51483dc29d..9f3178b309 100644 --- a/docs/specs/agent-environments/http-transport/verification.md +++ b/docs/specs/agent-environments/http-transport/verification.md @@ -281,3 +281,10 @@ An eleventh finding (Codex): URL construction canonicalized dot segments, so differs after URL parsing, so routing always sees the path the client sent (test: `/public/../mcp`, `/public/%2e%2e/mcp` and `/./mcp` get `400`; `/card?x=1` still routes). + +A twelfth finding (Codex): `bodyUsed` means disturbed, not consumed, so a +composed route that read one chunk of an upload left the rest paused on a +keep-alive socket. The close decision now uses the incoming message's end of +stream (test: a route that reads one chunk of a 256 KiB upload is answered +with `Connection: close` and a closed socket; a route that reads it all keeps +the connection). diff --git a/src/server/http/serveStreamableHttp.test.ts b/src/server/http/serveStreamableHttp.test.ts index 6e72d4d2ba..f5620af518 100644 --- a/src/server/http/serveStreamableHttp.test.ts +++ b/src/server/http/serveStreamableHttp.test.ts @@ -711,3 +711,54 @@ it("retries a shutdown hook that threw instead of marking it complete", async () await slot.close(); expect(attempts).toBe(2); }); + +it("closes the connection when a route read only part of the body and keeps it after a full read", async () => { + const { served } = await start({ + routes: [ + { + method: "POST", + path: "/partial", + handle: async (request) => { + const reader = request.body?.getReader(); + await reader?.read(); + return new Response("stopped early"); + }, + }, + { + method: "POST", + path: "/whole", + handle: async (request) => + Response.json({ bytes: (await request.text()).length }), + }, + ], + }); + const url = new URL(served.url); + const agent = new Agent({ keepAlive: true, maxSockets: 1 }); + const body = "x".repeat(256 * 1024); + try { + const partial = await keepAliveRequest(agent, new URL("/partial", url), { + method: "POST", + headers: { + "content-type": "text/plain", + "content-length": String(body.length), + }, + body, + }); + expect(partial.status).toBe(200); + expect(partial.headers.connection).toBe("close"); + expect(partial.socketClosed).toBe(true); + const whole = await keepAliveRequest(agent, new URL("/whole", url), { + method: "POST", + headers: { + "content-type": "text/plain", + "content-length": String(body.length), + }, + body, + }); + expect(whole.status).toBe(200); + expect(whole.headers.connection).not.toBe("close"); + expect(whole.socketClosed).toBe(false); + } finally { + agent.destroy(); + } +}); diff --git a/src/server/http/serveStreamableHttp.ts b/src/server/http/serveStreamableHttp.ts index 444cc6271d..837feadec4 100644 --- a/src/server/http/serveStreamableHttp.ts +++ b/src/server/http/serveStreamableHttp.ts @@ -398,15 +398,15 @@ const serveNodeRequest = async ( response: ServerResponse, ): Promise => { const { route, origin, logger } = target; - // A reply sent before an incoming body was read leaves that body on the socket, which the - // bridged stream keeps paused; closing the connection after the reply releases it. The - // incoming message decides, so a GET or HEAD that carried a body closes too. + // A reply sent before an incoming body was fully consumed leaves the rest on the socket, which + // the bridged stream keeps paused; closing the connection after the reply releases it. The + // incoming message decides (end of stream, not "body disturbed"), so a partially read upload + // and a GET or HEAD that carried a body close too. const declaresBody = request.headers["content-length"] !== undefined || request.headers["transfer-encoding"] !== undefined; let webRequest: Request | undefined; - const unreadBody = (): boolean => - declaresBody && !(webRequest?.bodyUsed ?? false); + const unreadBody = (): boolean => declaresBody && !request.readableEnded; try { try { webRequest = toWebRequest(request, response, origin());