diff --git a/src/request.ts b/src/request.ts index 49c18bb..c2ba1a5 100644 --- a/src/request.ts +++ b/src/request.ts @@ -261,6 +261,40 @@ const readBodyWithFastPath = ( }) } +const normalizeAbortError = ( + request: Record, + incoming: IncomingMessage | Http2ServerRequest +): Error => { + if (incoming.errored) { + return incoming.errored + } + const reason = request[abortReasonKey] + if (typeof reason !== 'undefined') { + return reason instanceof Error ? reason : new Error(String(reason)) + } + return new Error('Client connection prematurely closed.') +} + +// A client abort destroys the IncomingMessage, but does not clear its internal +// buffer. For HTTP/1, `complete === true` guarantees the parser received the +// entire message, so the unread body can still be drained synchronously. +// (Not valid for HTTP/2, where `complete` is also true for aborted requests. see https://nodejs.org/api/http2.html#requestcomplete) +const canBeRecoveredAfterAbort = ( + incoming: IncomingMessage | Http2ServerRequest +): incoming is IncomingMessage => { + return incoming.complete && incoming.httpVersionMajor === 1 +} + +const readIncomingMessageInternalBuffer = (incoming: IncomingMessage) => { + const chunks: Buffer[] = [] + let chunk: Buffer | string | null + while ((chunk = incoming.read()) !== null) { + chunks.push(Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk)) + } + const buffer = chunks.length === 1 ? chunks[0] : Buffer.concat(chunks) + return buffer +} + const readRawBodyIfAvailable = (request: Record): Buffer | undefined => { const incoming = request[incomingKey] as IncomingMessage | Http2ServerRequest if ('rawBody' in incoming && (incoming as any).rawBody instanceof Buffer) { @@ -283,7 +317,16 @@ const readBodyDirect = (request: Record): Promise const incoming = request[incomingKey] as IncomingMessage | Http2ServerRequest if (Readable.isDisturbed(incoming)) { - return rejectBodyUnusable() + if (incoming.readableDidRead) { + return rejectBodyUnusable() + } + // Aborted (e.g. client disconnect) before anything read the stream. + if (canBeRecoveredAfterAbort(incoming)) { + const buffer = readIncomingMessageInternalBuffer(incoming) + request[bodyBufferKey] = buffer + return Promise.resolve(buffer) + } + return Promise.reject(normalizeAbortError(request, incoming)) } const promise = new Promise((resolve, reject) => { @@ -319,17 +362,14 @@ const readBodyDirect = (request: Record): Promise onEnd() return } + if (canBeRecoveredAfterAbort(incoming)) { + // Drain what the abort left in the internal buffer. + readIncomingMessageInternalBuffer(incoming) + onEnd() + return + } finish(() => { - if (incoming.errored) { - reject(incoming.errored) - return - } - const reason = request[abortReasonKey] - if (reason !== undefined) { - reject(reason instanceof Error ? reason : new Error(String(reason))) - return - } - reject(new Error('Client connection prematurely closed.')) + reject(normalizeAbortError(request, incoming)) }) } const cleanup = () => { diff --git a/test/listener.test.ts b/test/listener.test.ts index 505db60..e9aecf4 100644 --- a/test/listener.test.ts +++ b/test/listener.test.ts @@ -1,5 +1,7 @@ import request from 'supertest' -import { createServer } from 'node:http' +import type { IncomingMessage } from 'node:http' +import { createServer, request as httpRequest } from 'node:http' +import type { AddressInfo } from 'node:net' import { getRequestListener } from '../src/listener' import { GlobalRequest, Request as LightweightRequest, RequestError } from '../src/request' import { GlobalResponse, Response as LightweightResponse } from '../src/response' @@ -422,6 +424,72 @@ describe('Abort request - error path', () => { ) }) +describe('Abort request - direct body read', () => { + // See https://github.com/honojs/node-server/issues/370 + const runClientDisconnectCase = async (clientBody: { + contentLength: number + write: string + }): Promise<{ body: Promise }> => { + // Wrapped in an object so `await` does not flatten the inner promise + let resolveBodyRead!: (result: { body: Promise }) => void + const bodyRead = new Promise<{ body: Promise }>((r) => { + resolveBodyRead = r + }) + + const requestListener = getRequestListener(async (req: Request, env) => { + const incoming = (env as { incoming: IncomingMessage }).incoming + // Read only after the client disconnect has destroyed the stream + if (!incoming.destroyed) { + await new Promise((resolve) => incoming.once('close', resolve)) + } + const body = req.text() + body.catch(() => {}) // inspected via expect(); avoid unhandled rejection + resolveBodyRead({ body }) + return new Response('ok') + }) + const server = createServer(requestListener) + await new Promise((resolve) => server.listen(0, resolve)) + const port = (server.address() as AddressInfo).port + + try { + const clientReq = httpRequest({ + port, + method: 'POST', + path: '/', + headers: { + 'content-type': 'text/plain', + 'content-length': clientBody.contentLength, + }, + }) + clientReq.on('error', () => {}) + clientReq.write(clientBody.write) + // Give the bytes time to reach the server's parser, then disconnect + await new Promise((resolve) => setTimeout(resolve, 30)) + clientReq.destroy() + return await withTimeout(bodyRead, 'handler did not read the body') + } finally { + server.close() + } + } + + it('should return the full body when the client disconnects after sending it', async () => { + const { body } = await runClientDisconnectCase({ + contentLength: Buffer.byteLength('hello world'), + write: 'hello world', + }) + await expect(body).resolves.toBe('hello world') + }) + + it('should reject with an abort error, not "Body is unusable", for a truncated body', async () => { + const { body } = await runClientDisconnectCase({ + contentLength: 100, + write: 'partial', + }) + await expect(body).rejects.toThrow(/aborted|prematurely closed/) + await expect(body).rejects.not.toThrow('Body is unusable') + }) +}) + describe('Abort request - cacheable response path', () => { it.each([ ['string', () => new Response('fast path')],