From 2cc04698d5c97218eb816d389222fb001dd8e4be Mon Sep 17 00:00:00 2001 From: Taku Amano Date: Tue, 14 Jul 2026 05:53:43 +0900 Subject: [PATCH 1/5] fix: decode string chunks with the stream encoding When application code has called setEncoding() on the IncomingMessage, data events yield strings in that encoding. Converting those chunks back to bytes as UTF-8 corrupts bodies decoded with another encoding. Decode with incoming.readableEncoding and cover the streamed latin1 case. --- src/request.ts | 8 +++++++- test/request.test.ts | 17 +++++++++++++++++ 2 files changed, 24 insertions(+), 1 deletion(-) diff --git a/src/request.ts b/src/request.ts index 49c18bb..2f050f9 100644 --- a/src/request.ts +++ b/src/request.ts @@ -57,6 +57,12 @@ const newHeadersFromIncoming = (incoming: IncomingMessage | Http2ServerRequest) export type IncomingMessageWithWrapBodyStream = IncomingMessage & { [wrapBodyStream]: boolean } export const wrapBodyStream = Symbol('wrapBodyStream') + +// When setEncoding() was called on the stream, chunks arrive as strings in +// that encoding — decode with it so the original bytes are reconstructed. +const toBufferChunk = (chunk: Buffer | string, encoding: BufferEncoding | null): Buffer => + Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk, encoding ?? 'utf8') + const newRequestFromIncoming = ( method: string, url: string, @@ -300,7 +306,7 @@ const readBodyDirect = (request: Record): Promise } const onData = (chunk: Buffer | string) => { - chunks.push(Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk)) + chunks.push(toBufferChunk(chunk, incoming.readableEncoding)) } const onEnd = () => { finish(() => { diff --git a/test/request.test.ts b/test/request.test.ts index e6abf26..748195f 100644 --- a/test/request.test.ts +++ b/test/request.test.ts @@ -177,6 +177,23 @@ describe('Request', () => { await expect(textPromise).rejects.toBeInstanceOf(Error) }) + it('should reconstruct the original bytes when setEncoding() was used on a streamed body', async () => { + const socket = new Socket() + const incomingMessage = new IncomingMessage(socket) + incomingMessage.method = 'POST' + incomingMessage.headers = { host: 'localhost' } + incomingMessage.rawHeaders = ['host', 'localhost'] + incomingMessage.url = '/foo.txt' + incomingMessage.setEncoding('latin1') + + const req = newRequest(incomingMessage) + const textPromise = req.text() + incomingMessage.push(Buffer.from('café', 'utf8')) + incomingMessage.push(null) + + await expect(textPromise).resolves.toBe('café') + }) + it('should reject direct body read for unsupported methods like native Request', async () => { for (const method of ['CONNECT', 'TRACK']) { const req = newRequest({ From 420efafe06e6e6c03b633f0ab7ab8f37faf1a2f1 Mon Sep 17 00:00:00 2001 From: Taku Amano Date: Tue, 14 Jul 2026 05:54:12 +0900 Subject: [PATCH 2/5] fix: recover complete HTTP/1 bodies after client disconnect A client can disconnect after Node has parsed the complete HTTP/1 request but before the application reads it. Recover the untouched buffered body for direct reads, native Request reads, formData(), and raw body streams. Share and cache the recovery result across paths, preserve application errors, reject lossy string encodings, and validate canonical content-length values. Add real-socket and synthetic coverage with bounded waits so regressions fail without leaking the test server. --- src/request.ts | 116 ++++++++++++++- test/request-client-disconnect.test.ts | 169 ++++++++++++++++++++++ test/request.test.ts | 191 +++++++++++++++++++++++++ 3 files changed, 473 insertions(+), 3 deletions(-) create mode 100644 test/request-client-disconnect.test.ts diff --git a/src/request.ts b/src/request.ts index 2f050f9..77df376 100644 --- a/src/request.ts +++ b/src/request.ts @@ -58,11 +58,96 @@ const newHeadersFromIncoming = (incoming: IncomingMessage | Http2ServerRequest) export type IncomingMessageWithWrapBodyStream = IncomingMessage & { [wrapBodyStream]: boolean } export const wrapBodyStream = Symbol('wrapBodyStream') +// Encodings whose decode → re-encode round-trip is byte-exact for any input. +// utf8/utf16le replace invalid sequences with U+FFFD and ascii masks the high +// bit of every byte, so a body decoded through them cannot be reconstructed +// reliably — recovery is refused for those instead of returning corrupt bytes. +const byteExactEncodings = new Set(['latin1', 'binary', 'hex', 'base64', 'base64url']) + +const isByteExactEncoding = (encoding: BufferEncoding | null): boolean => + encoding === null || byteExactEncodings.has(encoding) + +const bodyBufferedBeforeDisconnectKey = Symbol('bodyBufferedBeforeDisconnect') + // When setEncoding() was called on the stream, chunks arrive as strings in // that encoding — decode with it so the original bytes are reconstructed. const toBufferChunk = (chunk: Buffer | string, encoding: BufferEncoding | null): Buffer => Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk, encoding ?? 'utf8') +// A client may close the connection after Node has parsed the complete HTTP/1 +// request but before the application starts reading it. In that state the +// IncomingMessage is disturbed/aborted, while the untouched body is still in +// its internal readable buffer — recover it instead of failing the read. +// +// HTTP/2 is excluded deliberately: an RST_STREAM discards the buffered request +// data at the protocol layer, and Http2ServerRequest#complete is also true for +// aborted/destroyed streams, so it cannot be trusted as a "fully received" +// signal. +// +// Returns the buffered body, an Error the read must fail with, or undefined +// when the stream is not in the recoverable state. The result is cached on the +// incoming object so every read path observes the same outcome. +const readBodyBufferedBeforeDisconnect = ( + incoming: IncomingMessage | Http2ServerRequest +): Buffer | Error | undefined => { + if ( + incoming instanceof Http2ServerRequest || + !incoming.complete || + !incoming.readableAborted || + typeof incoming.read !== 'function' || + !isByteExactEncoding(incoming.readableEncoding) + ) { + return undefined + } + + const incomingWithCache = incoming as IncomingMessage & { + [bodyBufferedBeforeDisconnectKey]?: Buffer | Error + } + if (incomingWithCache[bodyBufferedBeforeDisconnectKey] !== undefined) { + return incomingWithCache[bodyBufferedBeforeDisconnectKey] + } + + let result: Buffer | Error + const errored = incoming.errored + if (errored && (errored as NodeJS.ErrnoException).code !== 'ECONNRESET') { + // The stream was destroyed with an application-provided error (e.g. a + // body-size guard calling incoming.destroy(err)). Node's own teardown of a + // disconnected client uses ECONNRESET errors, which must still recover the + // body — anything else is surfaced to the reader instead of swallowed. + result = errored + } else { + const chunk = incoming.read() as Buffer | string | null + const buffer = + chunk === null ? Buffer.alloc(0) : toBufferChunk(chunk, incoming.readableEncoding) + result = buffer + // Validate only canonical digit values: the HTTP parser guarantees this + // form on real connections, and lenient coercion of synthetic inputs + // (e.g. '0x10') would validate against a length the header never meant. + const contentLength = incoming.headers['content-length'] + if (typeof contentLength === 'string' && /^\d+$/.test(contentLength)) { + const expectedLength = Number(contentLength) + if (Number.isSafeInteger(expectedLength) && buffer.length !== expectedLength) { + result = newBodyUnusableError() + } + } + } + incomingWithCache[bodyBufferedBeforeDisconnectKey] = result + return result +} + +const enqueueBufferedBody = ( + controller: ReadableStreamDefaultController, + buffered: Buffer | Error +): void => { + if (buffered instanceof Error) { + controller.error(buffered) + return + } + if (buffered.length > 0) { + controller.enqueue(buffered) + } + controller.close() +} const newRequestFromIncoming = ( method: string, url: string, @@ -102,6 +187,13 @@ const newRequestFromIncoming = ( init.body = new ReadableStream({ async pull(controller) { try { + if (!reader) { + const buffered = readBodyBufferedBeforeDisconnect(incoming) + if (buffered !== undefined) { + enqueueBufferedBody(controller, buffered) + return + } + } reader ||= Readable.toWeb(incoming).getReader() const { done, value } = await reader.read() if (done) { @@ -115,8 +207,17 @@ const newRequestFromIncoming = ( }, }) } else { - // lazy-consume request body - init.body = Readable.toWeb(incoming) as ReadableStream + const buffered = readBodyBufferedBeforeDisconnect(incoming) + if (buffered !== undefined) { + init.body = new ReadableStream({ + start(controller) { + enqueueBufferedBody(controller, buffered) + }, + }) + } else { + // lazy-consume request body + init.body = Readable.toWeb(incoming) as ReadableStream + } } } @@ -288,10 +389,19 @@ const readBodyDirect = (request: Record): Promise } const incoming = request[incomingKey] as IncomingMessage | Http2ServerRequest - if (Readable.isDisturbed(incoming)) { + if (incoming.readableDidRead) { return rejectBodyUnusable() } + const buffered = readBodyBufferedBeforeDisconnect(incoming) + if (buffered !== undefined) { + if (buffered instanceof Error) { + return Promise.reject(buffered) + } + request[bodyBufferKey] = buffered + return Promise.resolve(buffered) + } + const promise = new Promise((resolve, reject) => { const chunks: Buffer[] = [] let settled = false diff --git a/test/request-client-disconnect.test.ts b/test/request-client-disconnect.test.ts new file mode 100644 index 0000000..6483b5c --- /dev/null +++ b/test/request-client-disconnect.test.ts @@ -0,0 +1,169 @@ +import type { Context } from 'hono' +import { Hono } from 'hono' +import { request as requestHTTP } from 'node:http' +import type { ClientRequest, OutgoingHttpHeaders } from 'node:http' +import type { AddressInfo } from 'node:net' +import { serve } from '../src/server' +import type { HttpBindings, ServerType } from '../src/types' + +type BodyReadResult = { body: unknown } | { error: unknown } +type BindingsContext = Context<{ Bindings: HttpBindings }> + +const closeServer = (server: ServerType): Promise => + new Promise((resolve, reject) => { + server.close((error?: Error) => { + if (error) { + reject(error) + } else { + resolve() + } + }) + }) + +// Every wait rejects after a deadline so a regression fails fast with a +// descriptive error instead of idling until the runner timeout — which would +// also skip the finally-based server cleanup and leak the listening handle. +const withTimeout = (promise: Promise, ms: number, label: string): Promise => { + let timer!: NodeJS.Timeout + const timeout = new Promise((_, reject) => { + timer = setTimeout(() => reject(new Error(`Timed out waiting for ${label}`)), ms) + }) + return Promise.race([promise, timeout]).finally(() => clearTimeout(timer)) +} + +const readBodyAfterDisconnect = async ( + body: string, + headers: OutgoingHttpHeaders, + send: (request: ClientRequest, body: string) => void, + read: (c: BindingsContext) => Promise = (c) => c.req.text() +): Promise => { + let notifyBodyComplete!: () => void + const bodyComplete = new Promise((resolve) => { + notifyBodyComplete = resolve + }) + + let reportResult!: (result: BodyReadResult) => void + const result = new Promise((resolve) => { + reportResult = resolve + }) + + const app = new Hono<{ Bindings: HttpBindings }>() + app.post('/', async (c) => { + try { + const incoming = c.env.incoming + const deadline = Date.now() + 1000 + while (!incoming.complete) { + if (Date.now() > deadline) { + throw new Error('Timed out waiting for the request body to complete') + } + await new Promise((resolve) => setImmediate(resolve)) + } + + const aborted = new Promise((resolve) => { + incoming.once('aborted', resolve) + }) + notifyBodyComplete() + await withTimeout(aborted, 1000, 'the client disconnect') + + reportResult({ body: await read(c) }) + } catch (error) { + reportResult({ error }) + } + return c.text('done') + }) + + let server!: ServerType + const address = await new Promise((resolve) => { + server = serve( + { + fetch: app.fetch, + hostname: '127.0.0.1', + port: 0, + }, + resolve + ) + }) + + const request = requestHTTP({ + headers, + host: address.address, + method: 'POST', + path: '/', + port: address.port, + }) + request.on('error', () => {}) + send(request, body) + + try { + await withTimeout(bodyComplete, 2000, 'the server to receive the request body') + request.destroy() + return await withTimeout(result, 4000, 'the body read result') + } finally { + request.destroy() + await closeServer(server) + } +} + +describe('body reads after client disconnect', () => { + const body = JSON.stringify({ hello: 'world' }) + + it('reads a complete content-length body', async () => { + const result = await readBodyAfterDisconnect( + body, + { 'content-length': Buffer.byteLength(body) }, + (request, value) => request.end(value) + ) + + expect(result).toEqual({ body }) + }) + + it('reads a complete chunked body', async () => { + const result = await readBodyAfterDisconnect(body, {}, (request, value) => { + request.write(value) + request.end() + }) + + expect(result).toEqual({ body }) + }) + + it('reads a complete body via formData()', async () => { + const form = 'hello=world&foo=bar' + const result = await readBodyAfterDisconnect( + form, + { + 'content-length': Buffer.byteLength(form), + 'content-type': 'application/x-www-form-urlencoded', + }, + (request, value) => request.end(value), + async (c) => Object.fromEntries((await c.req.formData()).entries()) + ) + + expect(result).toEqual({ body: { hello: 'world', foo: 'bar' } }) + }) + + it('reads a complete body through the raw body stream', async () => { + const result = await readBodyAfterDisconnect( + body, + { 'content-length': Buffer.byteLength(body) }, + (request, value) => request.end(value), + (c) => new Response(c.req.raw.body).text() + ) + + expect(result).toEqual({ body }) + }) + + it('reads a complete body even when request properties were accessed after disconnect', async () => { + const result = await readBodyAfterDisconnect( + body, + { 'content-length': Buffer.byteLength(body) }, + (request, value) => request.end(value), + (c) => { + // creates the internal native Request cache before the body is read + void c.req.raw.cache + return c.req.text() + } + ) + + expect(result).toEqual({ body }) + }) +}) diff --git a/test/request.test.ts b/test/request.test.ts index 748195f..94cd5c6 100644 --- a/test/request.test.ts +++ b/test/request.test.ts @@ -177,6 +177,197 @@ describe('Request', () => { await expect(textPromise).rejects.toBeInstanceOf(Error) }) + it('should preserve the connection error when an incomplete stream was destroyed before reading', async () => { + const socket = new Socket() + const incomingMessage = new IncomingMessage(socket) + incomingMessage.method = 'POST' + incomingMessage.headers = { + host: 'localhost', + 'content-length': '6', + } + incomingMessage.rawHeaders = ['host', 'localhost', 'content-length', '6'] + incomingMessage.url = '/foo.txt' + incomingMessage.push('foo') + incomingMessage.on('error', () => {}) + const req = newRequest(incomingMessage) + + const closed = new Promise((resolve) => { + incomingMessage.once('close', resolve) + }) + incomingMessage.destroy(new Error('aborted')) + await closed + + await expect(req.text()).rejects.toThrow('aborted') + }) + + const newDestroyedIncomingWithCompleteBody = async ( + body: string, + headers: Record = {}, + destroyError?: Error + ): Promise => { + const socket = new Socket() + const incomingMessage = new IncomingMessage(socket) + incomingMessage.method = 'POST' + incomingMessage.headers = { host: 'localhost', ...headers } + incomingMessage.rawHeaders = ['host', 'localhost', ...Object.entries(headers).flat()] + incomingMessage.url = '/foo.txt' + incomingMessage.push(body) + incomingMessage.complete = true + incomingMessage.on('error', () => {}) + + const closed = new Promise((resolve) => { + incomingMessage.once('close', resolve) + }) + incomingMessage.destroy(destroyError) + await closed + return incomingMessage + } + + it('should preserve the complete buffered body when the stream was destroyed before reading', async () => { + const incomingMessage = await newDestroyedIncomingWithCompleteBody('foo', { + 'content-length': '3', + }) + const req = newRequest(incomingMessage) + + await expect(req.text()).resolves.toBe('foo') + }) + + it('should reject when the buffered body does not match content-length', async () => { + const incomingMessage = await newDestroyedIncomingWithCompleteBody('foo', { + 'content-length': '6', + }) + const req = newRequest(incomingMessage) + + await expect(req.text()).rejects.toThrow('Body is unusable') + }) + + it('should preserve the complete buffered body for standard path reads like formData()', async () => { + const incomingMessage = await newDestroyedIncomingWithCompleteBody('hello=world&foo=bar', { + 'content-type': 'application/x-www-form-urlencoded', + 'content-length': '19', + }) + const req = newRequest(incomingMessage) + + const formData = await req.formData() + expect(Object.fromEntries(formData.entries())).toEqual({ hello: 'world', foo: 'bar' }) + }) + + it('should preserve the complete buffered body when read through the body stream', async () => { + const incomingMessage = await newDestroyedIncomingWithCompleteBody('foo', { + 'content-length': '3', + }) + const req = newRequest(incomingMessage) + + await expect(new Response(req.body).text()).resolves.toBe('foo') + }) + + it('should reject with the underlying error when a complete stream was destroyed with an application error', async () => { + const incomingMessage = await newDestroyedIncomingWithCompleteBody( + 'foo', + { 'content-length': '3' }, + new Error('Payload too large') + ) + const req = newRequest(incomingMessage) + + await expect(req.text()).rejects.toThrow('Payload too large') + }) + + it('should reject standard path reads when a complete stream was destroyed with an application error', async () => { + const incomingMessage = await newDestroyedIncomingWithCompleteBody( + 'hello=world', + { + 'content-type': 'application/x-www-form-urlencoded', + 'content-length': '11', + }, + new Error('Payload too large') + ) + const req = newRequest(incomingMessage) + + await expect(req.formData()).rejects.toThrow('Payload too large') + }) + + it('should preserve the complete buffered body when destroyed by a client disconnect error', async () => { + const incomingMessage = await newDestroyedIncomingWithCompleteBody( + 'foo', + { 'content-length': '3' }, + Object.assign(new Error('aborted'), { code: 'ECONNRESET' }) + ) + const req = newRequest(incomingMessage) + + await expect(req.text()).resolves.toBe('foo') + }) + + it('should preserve the complete buffered body when request properties were accessed after disconnect', async () => { + const incomingMessage = await newDestroyedIncomingWithCompleteBody('foo', { + 'content-length': '3', + }) + const req = newRequest(incomingMessage) + + // creates the internal native Request cache before the body is read + void req.cache + + await expect(req.text()).resolves.toBe('foo') + }) + + it('should skip content-length validation for non-canonical header values', async () => { + // The HTTP parser guarantees digit-only content-length on real + // connections; synthetic values must not be coerced into a bogus length. + for (const contentLength of ['0x10', '10, 10', '-3']) { + const incomingMessage = await newDestroyedIncomingWithCompleteBody('foo', { + 'content-length': contentLength, + }) + const req = newRequest(incomingMessage) + + await expect(req.text()).resolves.toBe('foo') + } + }) + + it('should reconstruct the original bytes when setEncoding() was used before disconnect', async () => { + const socket = new Socket() + const incomingMessage = new IncomingMessage(socket) + incomingMessage.method = 'POST' + incomingMessage.headers = { host: 'localhost' } + incomingMessage.rawHeaders = ['host', 'localhost'] + incomingMessage.url = '/foo.txt' + incomingMessage.setEncoding('latin1') + incomingMessage.push(Buffer.from('café', 'utf8')) + incomingMessage.complete = true + incomingMessage.on('error', () => {}) + + const closed = new Promise((resolve) => { + incomingMessage.once('close', resolve) + }) + incomingMessage.destroy() + await closed + + const req = newRequest(incomingMessage) + await expect(req.text()).resolves.toBe('café') + }) + + it('should not recover a body buffered with a lossy encoding', async () => { + for (const encoding of ['ascii', 'utf8'] as const) { + const socket = new Socket() + const incomingMessage = new IncomingMessage(socket) + incomingMessage.method = 'POST' + incomingMessage.headers = { host: 'localhost' } + incomingMessage.rawHeaders = ['host', 'localhost'] + incomingMessage.url = '/foo.txt' + incomingMessage.setEncoding(encoding) + incomingMessage.push(Buffer.from([0x61, 0xe9, 0x62])) + incomingMessage.complete = true + incomingMessage.on('error', () => {}) + + const closed = new Promise((resolve) => { + incomingMessage.once('close', resolve) + }) + incomingMessage.destroy() + await closed + + const req = newRequest(incomingMessage) + await expect(req.text()).rejects.toThrow('Client connection prematurely closed.') + } + }) + it('should reconstruct the original bytes when setEncoding() was used on a streamed body', async () => { const socket = new Socket() const incomingMessage = new IncomingMessage(socket) From 138386d309212c36bc9d4888a37de5bfba766e99 Mon Sep 17 00:00:00 2001 From: Taku Amano Date: Mon, 13 Jul 2026 18:10:20 +0900 Subject: [PATCH 3/5] fix: avoid recovering partially consumed request bodies Inspired by https://github.com/honojs/node-server/pull/373. Co-authored-by: BlankParticle --- src/listener.ts | 3 + src/request.ts | 100 +++++++++++++++++++++---- test/request-client-disconnect.test.ts | 20 +++++ test/request.test.ts | 63 +++++++++++++++- 4 files changed, 167 insertions(+), 19 deletions(-) diff --git a/src/listener.ts b/src/listener.ts index 3c4bea7..ae2991c 100644 --- a/src/listener.ts +++ b/src/listener.ts @@ -6,6 +6,7 @@ import type { IncomingMessageWithWrapBodyStream } from './request' import { abortRequest, newRequest, + recordBodyBufferedBeforeDisconnect, Request as LightweightRequest, wrapBodyStream, toRequestError, @@ -95,8 +96,10 @@ const makeCloseHandler = ): (() => void) => () => { if (incoming.errored) { + recordBodyBufferedBeforeDisconnect(incoming) req[abortRequest](incoming.errored.toString()) } else if (!outgoing.writableFinished) { + recordBodyBufferedBeforeDisconnect(incoming) req[abortRequest]('Client connection prematurely closed.') } diff --git a/src/request.ts b/src/request.ts index 77df376..4d1eae7 100644 --- a/src/request.ts +++ b/src/request.ts @@ -68,12 +68,42 @@ const isByteExactEncoding = (encoding: BufferEncoding | null): boolean => encoding === null || byteExactEncodings.has(encoding) const bodyBufferedBeforeDisconnectKey = Symbol('bodyBufferedBeforeDisconnect') +const bodyBufferedLengthBeforeDisconnectKey = Symbol('bodyBufferedLengthBeforeDisconnect') + +type IncomingWithBodyRecovery = IncomingMessage & { + [bodyBufferedBeforeDisconnectKey]?: Buffer | Error + [bodyBufferedLengthBeforeDisconnectKey]?: number +} // When setEncoding() was called on the stream, chunks arrive as strings in // that encoding — decode with it so the original bytes are reconstructed. const toBufferChunk = (chunk: Buffer | string, encoding: BufferEncoding | null): Buffer => Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk, encoding ?? 'utf8') +const isRecoverableDisconnectedIncoming = ( + incoming: IncomingMessage | Http2ServerRequest +): incoming is IncomingMessage => + !(incoming instanceof Http2ServerRequest) && + !!incoming.complete && + !!incoming.readableAborted && + typeof incoming.read === 'function' && + isByteExactEncoding(incoming.readableEncoding) + +// Remember how much complete HTTP/1 body data remained buffered when the +// disconnect was observed. A later raw read can drain a destroyed stream +// without updating readableDidRead, so the length snapshot is needed to keep +// the Fetch-style body from silently resolving with only the remainder. +export const recordBodyBufferedBeforeDisconnect = ( + incoming: IncomingMessage | Http2ServerRequest +): void => { + if (incoming.readableDidRead || !isRecoverableDisconnectedIncoming(incoming)) { + return + } + + const incomingWithRecovery = incoming as IncomingWithBodyRecovery + incomingWithRecovery[bodyBufferedLengthBeforeDisconnectKey] ??= incoming.readableLength +} + // A client may close the connection after Node has parsed the complete HTTP/1 // request but before the application starts reading it. In that state the // IncomingMessage is disturbed/aborted, while the untouched body is still in @@ -88,23 +118,16 @@ const toBufferChunk = (chunk: Buffer | string, encoding: BufferEncoding | null): // when the stream is not in the recoverable state. The result is cached on the // incoming object so every read path observes the same outcome. const readBodyBufferedBeforeDisconnect = ( - incoming: IncomingMessage | Http2ServerRequest + incoming: IncomingMessage | Http2ServerRequest, + chunks?: Buffer[] ): Buffer | Error | undefined => { - if ( - incoming instanceof Http2ServerRequest || - !incoming.complete || - !incoming.readableAborted || - typeof incoming.read !== 'function' || - !isByteExactEncoding(incoming.readableEncoding) - ) { + if ((incoming.readableDidRead && !chunks) || !isRecoverableDisconnectedIncoming(incoming)) { return undefined } - const incomingWithCache = incoming as IncomingMessage & { - [bodyBufferedBeforeDisconnectKey]?: Buffer | Error - } - if (incomingWithCache[bodyBufferedBeforeDisconnectKey] !== undefined) { - return incomingWithCache[bodyBufferedBeforeDisconnectKey] + const incomingWithRecovery = incoming as IncomingWithBodyRecovery + if (incomingWithRecovery[bodyBufferedBeforeDisconnectKey] !== undefined) { + return incomingWithRecovery[bodyBufferedBeforeDisconnectKey] } let result: Buffer | Error @@ -115,10 +138,18 @@ const readBodyBufferedBeforeDisconnect = ( // disconnected client uses ECONNRESET errors, which must still recover the // body — anything else is surfaced to the reader instead of swallowed. result = errored + } else if ( + incomingWithRecovery[bodyBufferedLengthBeforeDisconnectKey] !== undefined && + incoming.readableLength !== incomingWithRecovery[bodyBufferedLengthBeforeDisconnectKey] + ) { + result = newBodyUnusableError() } else { + const bodyChunks = chunks ?? [] const chunk = incoming.read() as Buffer | string | null - const buffer = - chunk === null ? Buffer.alloc(0) : toBufferChunk(chunk, incoming.readableEncoding) + if (chunk !== null) { + bodyChunks.push(toBufferChunk(chunk, incoming.readableEncoding)) + } + const buffer = bodyChunks.length === 1 ? bodyChunks[0] : Buffer.concat(bodyChunks) result = buffer // Validate only canonical digit values: the HTTP parser guarantees this // form on real connections, and lenient coercion of synthetic inputs @@ -131,7 +162,7 @@ const readBodyBufferedBeforeDisconnect = ( } } } - incomingWithCache[bodyBufferedBeforeDisconnectKey] = result + incomingWithRecovery[bodyBufferedBeforeDisconnectKey] = result return result } @@ -415,6 +446,37 @@ const readBodyDirect = (request: Record): Promise callback() } + // readBodyDirect started while the stream was untouched, so it owns every + // chunk emitted from that point onward. If HTTP/1 parsing completed before + // the connection closed, combine those chunks with anything still buffered + // and treat the body as complete. Transport-level ECONNRESET is recoverable; + // application-provided stream errors are not. + const recoverCompleteBodyAfterDisconnect = (error?: unknown): boolean => { + const streamError = incoming.errored ?? error + if ( + incoming instanceof Http2ServerRequest || + !incoming.complete || + !incoming.readableAborted || + typeof incoming.read !== 'function' || + (streamError && (streamError as NodeJS.ErrnoException).code !== 'ECONNRESET') + ) { + return false + } + + finish(() => { + const recovered = readBodyBufferedBeforeDisconnect(incoming, chunks) + if (recovered instanceof Error) { + reject(recovered) + } else if (recovered === undefined) { + reject(error ?? new Error('Client connection prematurely closed.')) + } else { + request[bodyBufferKey] = recovered + resolve(recovered) + } + }) + return true + } + const onData = (chunk: Buffer | string) => { chunks.push(toBufferChunk(chunk, incoming.readableEncoding)) } @@ -426,6 +488,9 @@ const readBodyDirect = (request: Record): Promise }) } const onError = (error: unknown) => { + if (recoverCompleteBodyAfterDisconnect(error)) { + return + } finish(() => { reject(error) }) @@ -435,6 +500,9 @@ const readBodyDirect = (request: Record): Promise onEnd() return } + if (recoverCompleteBodyAfterDisconnect()) { + return + } finish(() => { if (incoming.errored) { reject(incoming.errored) diff --git a/test/request-client-disconnect.test.ts b/test/request-client-disconnect.test.ts index 6483b5c..2167297 100644 --- a/test/request-client-disconnect.test.ts +++ b/test/request-client-disconnect.test.ts @@ -126,6 +126,26 @@ describe('body reads after client disconnect', () => { expect(result).toEqual({ body }) }) + it('rejects a body read after the raw stream consumed the disconnected body', async () => { + let rawBody: string | undefined + const result = await readBodyAfterDisconnect( + body, + {}, + (request, value) => { + request.write(value) + request.end() + }, + (c) => { + rawBody = c.env.incoming.read()?.toString() + return c.req.text() + } + ) + + expect(rawBody).toBe(body) + expect(result).toEqual({ error: expect.any(TypeError) }) + expect((result as { error: Error }).error.message).toBe('Body is unusable') + }) + it('reads a complete body via formData()', async () => { const form = 'hello=world&foo=bar' const result = await readBodyAfterDisconnect( diff --git a/test/request.test.ts b/test/request.test.ts index 94cd5c6..e00e17c 100644 --- a/test/request.test.ts +++ b/test/request.test.ts @@ -143,7 +143,7 @@ describe('Request', () => { expect(req.signal.aborted).toBe(true) }) - it('should reject direct body read when incoming stream is destroyed mid-read', async () => { + it('should preserve an application error when a complete stream is destroyed mid-read', async () => { const socket = new Socket() const incomingMessage = new IncomingMessage(socket) incomingMessage.method = 'POST' @@ -155,11 +155,44 @@ describe('Request', () => { const req = newRequest(incomingMessage) const textPromise = req.json() - incomingMessage.destroy(new Error('Client connection prematurely closed.')) + const error = new Error('Payload too large') + incomingMessage.complete = true + incomingMessage.destroy(error) - await expect(textPromise).rejects.toBeInstanceOf(Error) + await expect(textPromise).rejects.toBe(error) }) + it.each([ + ['without a stream error', undefined], + [ + 'with a client disconnect error', + Object.assign(new Error('aborted'), { code: 'ECONNRESET' }), + ], + ])( + 'should preserve a complete body when the stream closes mid-read %s', + async (_label, destroyError) => { + const socket = new Socket() + const incomingMessage = new IncomingMessage(socket) + incomingMessage.method = 'POST' + incomingMessage.headers = { + host: 'localhost', + 'content-length': '6', + } + incomingMessage.rawHeaders = ['host', 'localhost', 'content-length', '6'] + incomingMessage.url = '/foo.txt' + const req = newRequest(incomingMessage) + + const textPromise = req.text() + incomingMessage.push('foo') + incomingMessage.pause() + incomingMessage.push('bar') + incomingMessage.complete = true + incomingMessage.destroy(destroyError) + + await expect(textPromise).resolves.toBe('foobar') + } + ) + it('should reject direct body read when incoming stream is destroyed without error', async () => { const socket = new Socket() const incomingMessage = new IncomingMessage(socket) @@ -261,6 +294,30 @@ describe('Request', () => { await expect(new Response(req.body).text()).resolves.toBe('foo') }) + it('should not recover only the remaining chunked body after a partial read', async () => { + const socket = new Socket() + const incomingMessage = new IncomingMessage(socket) + incomingMessage.method = 'POST' + incomingMessage.headers = { host: 'localhost', 'transfer-encoding': 'chunked' } + incomingMessage.rawHeaders = ['host', 'localhost', 'transfer-encoding', 'chunked'] + incomingMessage.url = '/foo.txt' + incomingMessage.push('first') + expect(incomingMessage.read()).toEqual(Buffer.from('first')) + expect(incomingMessage.readableDidRead).toBe(true) + incomingMessage.push('second') + incomingMessage.complete = true + incomingMessage.on('error', () => {}) + + const closed = new Promise((resolve) => { + incomingMessage.once('close', resolve) + }) + incomingMessage.destroy() + await closed + + const req = newRequest(incomingMessage) + expect(() => req.body).toThrow(TypeError) + }) + it('should reject with the underlying error when a complete stream was destroyed with an application error', async () => { const incomingMessage = await newDestroyedIncomingWithCompleteBody( 'foo', From 0007a09d1e8465bf12cb7488571470af7aa6788d Mon Sep 17 00:00:00 2001 From: Taku Amano Date: Mon, 13 Jul 2026 19:12:30 +0900 Subject: [PATCH 4/5] refactor: extract normalizeAbortError for body read failures Adopts the normalizeAbortError helper from https://github.com/honojs/node-server/pull/373. Co-authored-by: BlankParticle --- src/request.ts | 30 +++++++++++++++++++----------- 1 file changed, 19 insertions(+), 11 deletions(-) diff --git a/src/request.ts b/src/request.ts index 4d1eae7..0e019c9 100644 --- a/src/request.ts +++ b/src/request.ts @@ -407,6 +407,23 @@ const readRawBodyIfAvailable = (request: Record): Buffer | return undefined } +// The error a body read fails with after an abort: the stream's own error if +// it has one, otherwise the abort reason recorded on the request, otherwise a +// generic disconnect error. +const normalizeAbortError = ( + request: Record, + incoming: IncomingMessage | Http2ServerRequest +): Error => { + if (incoming.errored) { + return incoming.errored + } + const reason = request[abortReasonKey] + if (reason !== undefined) { + return reason instanceof Error ? reason : new Error(String(reason)) + } + return new Error('Client connection prematurely closed.') +} + // Read body directly from the IncomingMessage stream, bypassing Request object creation. // Precondition: the caller (listener.ts) must ensure that the IncomingMessage stream is // properly cleaned up (e.g. via incoming.resume()) when the response ends or the connection @@ -468,7 +485,7 @@ const readBodyDirect = (request: Record): Promise if (recovered instanceof Error) { reject(recovered) } else if (recovered === undefined) { - reject(error ?? new Error('Client connection prematurely closed.')) + reject(error ?? normalizeAbortError(request, incoming)) } else { request[bodyBufferKey] = recovered resolve(recovered) @@ -504,16 +521,7 @@ const readBodyDirect = (request: Record): Promise 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 = () => { From c86c3e11869615b4d81edcb26916fdec47939ad1 Mon Sep 17 00:00:00 2001 From: Taku Amano Date: Tue, 14 Jul 2026 21:30:52 +0900 Subject: [PATCH 5/5] refactor: reuse disconnected body recovery guard Co-authored-by: BlankParticle --- src/request.ts | 5 +---- 1 file changed, 1 insertion(+), 4 deletions(-) diff --git a/src/request.ts b/src/request.ts index 0e019c9..e963fd9 100644 --- a/src/request.ts +++ b/src/request.ts @@ -471,10 +471,7 @@ const readBodyDirect = (request: Record): Promise const recoverCompleteBodyAfterDisconnect = (error?: unknown): boolean => { const streamError = incoming.errored ?? error if ( - incoming instanceof Http2ServerRequest || - !incoming.complete || - !incoming.readableAborted || - typeof incoming.read !== 'function' || + !isRecoverableDisconnectedIncoming(incoming) || (streamError && (streamError as NodeJS.ErrnoException).code !== 'ECONNRESET') ) { return false