diff --git a/src/listener.ts b/src/listener.ts index c8c2763..b7d15bb 100644 --- a/src/listener.ts +++ b/src/listener.ts @@ -5,6 +5,8 @@ import type { Writable } from 'node:stream' import type { IncomingMessageWithWrapBodyStream } from './request' import { abortRequest, + bodySharedBufferKey, + releaseSharedBody, newRequest, recordBodyBufferedBeforeDisconnect, Request as LightweightRequest, @@ -99,6 +101,8 @@ const makeCloseHandler = needsBodyCleanup: boolean ): (() => void) => () => { + // The response is over: release clone-shared body buffers. + req[releaseSharedBody]() if (incoming.errored) { recordBodyBufferedBeforeDisconnect(incoming) req[abortRequest](incoming.errored.toString()) @@ -431,6 +435,16 @@ export const getRequestListener = ( // Synchronous cacheable response — no close listener needed. // No I/O events can fire between fetchCallback returning and responseViaCache // completing, so abort detection is not needed here. + // Release the clone-shared body buffer once the response completes, but + // only when there is one: the fast path must stay free of close + // listeners otherwise. Hook 'close' rather than 'finish' so the buffer + // is also released when the client disconnects before the response is + // flushed. + if (req[bodySharedBufferKey]) { + outgoing.once('close', () => { + req[releaseSharedBody]() + }) + } if (needsBodyCleanup && !incoming.readableEnded) { // Handler returned without consuming the body; drain after the // response is flushed so the socket is freed gracefully (avoids diff --git a/src/request.ts b/src/request.ts index a4b0e11..a2e0c9a 100644 --- a/src/request.ts +++ b/src/request.ts @@ -259,6 +259,15 @@ const bodyReadPromiseKey = Symbol('bodyReadPromise') const bodyConsumedDirectlyKey = Symbol('bodyConsumedDirectly') const bodyLockReaderKey = Symbol('bodyLockReader') const abortReasonKey = Symbol('abortReason') +// clone() switches the request to buffer-once + replay semantics: the body is +// read from the IncomingMessage exactly once into a Buffer shared by the +// original request and every clone, instead of tee()-ing the socket-backed +// stream, whose unread branch pins the whole raw body in memory for as long as +// the request graph stays reachable. See https://github.com/honojs/node-server/issues/347 +export const bodySharedBufferKey = Symbol('bodySharedBuffer') +const bodySharedStreamsKey = Symbol('bodySharedStreams') +const bodyReleasedKey = Symbol('bodyReleased') +export const releaseSharedBody = Symbol('releaseSharedBody') const newBodyUnusableError = (): TypeError => { return new TypeError('Body is unusable') @@ -268,6 +277,23 @@ const rejectBodyUnusable = (): Promise => { return Promise.reject(newBodyUnusableError()) } +const createBufferReplayStream = ( + bufferPromise: Promise, + streams: Set> +): ReadableStream => { + const stream = new ReadableStream({ + async start(controller) { + try { + enqueueBufferedBody(controller, await bufferPromise) + } catch (error) { + controller.error(error) + } + }, + }) + streams.add(stream) + return stream +} + const textDecoder = new TextDecoder() const consumeBodyDirectOnce = ( @@ -549,6 +575,44 @@ const readBodyDirect = (request: Record): Promise return promise } +// Read the body exactly once into the Buffer shared by the original request +// and its clones. Cache the resolved Buffer for later direct reads of the +// original request, unless the response already completed and the buffer +// was released. +const shareBodyBufferOnce = (request: Record): Promise => { + if (!request[bodySharedBufferKey]) { + const raw = readRawBodyIfAvailable(request) + request[bodySharedBufferKey] = raw ? Promise.resolve(raw) : readBodyDirect(request) + } + const bufferPromise = request[bodySharedBufferKey] as Promise + bufferPromise.then( + (buffer) => { + if (!request[bodyReleasedKey]) { + request[bodyBufferKey] ||= buffer + } + }, + () => {} // readers surface the error via their replay stream + ) + return bufferPromise +} + +const newReplayRequest = ( + request: Record, + method: string, + bufferPromise: Promise +): Request => { + const replayStreams = (request[bodySharedStreamsKey] ??= new Set>()) + return new Request( + request[urlKey] as string, + { + method, + headers: request.headers, + signal: request[getAbortController]().signal, + body: createBufferReplayStream(bufferPromise, replayStreams), + } as RequestInit + ) +} + const requestPrototype: Record = { get method() { return this[methodKey] @@ -580,6 +644,22 @@ const requestPrototype: Record = { return this[abortControllerKey] }, + // Called by the listener when the response has completed: cancel replay + // streams nobody is reading and drop the shared body buffer, so a retained + // request no longer pins the raw body in memory. Locked streams (actively + // being read) are left alone and finish naturally. + [releaseSharedBody]() { + this[bodyReleasedKey] = true + for (const stream of this[bodySharedStreamsKey] ?? []) { + if (!stream.locked) { + stream.cancel().catch(() => {}) + } + } + this[bodySharedStreamsKey] = undefined + this[bodySharedBufferKey] = undefined + this[bodyBufferKey] = undefined + }, + [getRequestCache]() { const abortController = this[getAbortController]() if (this[requestCache]) { @@ -616,6 +696,10 @@ const requestPrototype: Record = { return (this[requestCache] = req) } + if (this[bodySharedBufferKey]) { + return (this[requestCache] = newReplayRequest(this, method, this[bodySharedBufferKey])) + } + return (this[requestCache] = newRequestFromIncoming( this.method, this[urlKey], @@ -673,18 +757,30 @@ Object.defineProperty(requestPrototype, 'signal', { }, }) }) -;['clone', 'formData'].forEach((k) => { - Object.defineProperty(requestPrototype, k, { - value: function () { - if (this[bodyConsumedDirectlyKey]) { - if (k === 'clone') { - throw newBodyUnusableError() - } - return rejectBodyUnusable() - } - return this[getRequestCache]()[k]() - }, - }) +Object.defineProperty(requestPrototype, 'formData', { + value: function () { + if (this[bodyConsumedDirectlyKey]) { + return rejectBodyUnusable() + } + return this[getRequestCache]().formData() + }, +}) +Object.defineProperty(requestPrototype, 'clone', { + value: function (): Request { + if (this[bodyConsumedDirectlyKey]) { + throw newBodyUnusableError() + } + const method = this.method as string + if (method === 'GET' || method === 'HEAD' || method === 'TRACE') { + return this[getRequestCache]().clone() + } + // A cached native Request already owns the socket-backed body; its native + // clone() (tee) keeps the socket single-consumer. + if (this[requestCache]) { + return (this[requestCache] as Request).clone() + } + return newReplayRequest(this, method, shareBodyBufferOnce(this)) + }, }) // Direct body reading for text/arrayBuffer/blob/json: bypass getRequestCache() diff --git a/test/listener.test.ts b/test/listener.test.ts index 14a7690..aa337af 100644 --- a/test/listener.test.ts +++ b/test/listener.test.ts @@ -627,6 +627,81 @@ describe('Non-standard incoming request', () => { }) }) +describe('Release cloned request body', () => { + it('should keep the original body readable after a clone has consumed it', async () => { + const body = JSON.stringify({ data: 'foobar' }) + const requestListener = getRequestListener(async (req) => { + const { data } = (await req.clone().json()) as { data: string } + // middleware read the clone; the handler still reads the original body, + // like a native Request allows after clone() + return new Response(`${data}:${await req.text()}`) + }) + const server = createServer(requestListener) + + const res = await requestServer(server, { + method: 'POST', + path: '/', + headers: { 'content-type': 'application/json' }, + body, + }) + expect(res.status).toBe(200) + expect(await res.text()).toBe(`foobar:${body}`) + }) + + it('should release the shared body of a synchronous cacheable response on close', async () => { + class MockIncomingMessage extends Readable { + method = 'POST' + url = '/' + headers = { host: 'localhost' } + rawHeaders = ['host', 'localhost'] + + constructor() { + super() + this.push('foobar') + this.push(null) + } + + _read() { + // The full body is pushed in the constructor; nothing to pull. + } + } + + class MockServerResponse extends EventEmitter { + headersSent = false + writableFinished = false + + writeHead() { + this.headersSent = true + return this + } + + end() { + this.writableFinished = true + this.emit('finish') + this.emit('close') + return this + } + } + + let unreadClone: LightweightRequest | undefined + const requestListener = getRequestListener((req) => { + unreadClone = req.clone() + return new Response('fast path') + }) + + await requestListener( + new MockIncomingMessage() as unknown as IncomingMessage, + new MockServerResponse() as unknown as ServerResponse + ) + + // 'close' has been emitted: the unread clone no longer pins the body — + // its canceled replay stream is disturbed, like a native Request whose + // body stream was canceled + expect(unreadClone).toBeDefined() + await expect(unreadClone!.text()).rejects.toThrow(TypeError) + }) +}) + describe('overrideGlobalObjects', () => { const fetchCallback = vi.fn() diff --git a/test/request.test.ts b/test/request.test.ts index e00e17c..c7ba36f 100644 --- a/test/request.test.ts +++ b/test/request.test.ts @@ -10,6 +10,7 @@ import { GlobalRequest, getAbortController, abortControllerKey, + releaseSharedBody, RequestError, } from '../src/request' @@ -561,6 +562,72 @@ describe('Request', () => { expect(formData.get('b')).toBe('2') }) + it('should replay the buffered body to clones while leaving the original readable', 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.push('foobar') + incomingMessage.push(null) + const req = newRequest(incomingMessage) + + const clone1 = req.clone() + const clone2 = req.clone() + expect(req.bodyUsed).toBe(false) + await expect(clone1.text()).resolves.toBe('foobar') + await expect(clone2.text()).resolves.toBe('foobar') + // the clones replay the body instead of consuming it, so the original + // request stays readable + await expect(req.text()).resolves.toBe('foobar') + }) + + it('should cancel unread clones when the shared body buffer is released', 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.push('foobar') + incomingMessage.push(null) + const req = newRequest(incomingMessage) + + const clone = req.clone() + req[releaseSharedBody]() + // a clone nobody is reading no longer delivers — or pins — the body: + // canceling its replay stream marks it disturbed, like a native Request + // whose body stream was canceled + await expect(clone.text()).rejects.toThrow(TypeError) + }) + + it('should keep delivering to a clone that is being read when the shared body buffer is released', 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.push('foobar') + incomingMessage.push(null) + const req = newRequest(incomingMessage) + + const clone = req.clone() + const reader = clone.body!.getReader() + req[releaseSharedBody]() + const { done, value } = await reader.read() + expect(done).toBe(false) + expect(Buffer.from(value as Uint8Array).toString()).toBe('foobar') + await expect(reader.read()).resolves.toMatchObject({ done: true }) + }) + it('should reject direct body read when incoming stream has already been consumed', async () => { const socket = new Socket() const incomingMessage = new IncomingMessage(socket) diff --git a/test/server.test.ts b/test/server.test.ts index 387353d..33c7968 100644 --- a/test/server.test.ts +++ b/test/server.test.ts @@ -9,7 +9,12 @@ import { createServer as createHttp2Server } from 'node:http2' import { createServer as createHTTPSServer } from 'node:https' import { gunzipSync, inflateSync } from 'node:zlib' import { GlobalHeaders } from '../src/headers' -import { GlobalRequest, Request as LightweightRequest, getAbortController } from '../src/request' +import { + GlobalRequest, + Request as LightweightRequest, + getAbortController, + bodySharedBufferKey, +} from '../src/request' import { GlobalResponse, Response as LightweightResponse } from '../src/response' import { createAdaptorServer, serve } from '../src/server' import type { HttpBindings, ServerType } from '../src/types' @@ -1236,6 +1241,64 @@ describe('Memory leak test', () => { }) }) +describe('Memory leak test - cloned request body', () => { + let counter = 0 + const registry = new FinalizationRegistry(() => { + counter-- + }) + // Simulates production retention (logger/tracing contexts) that outlives the + // response. + const retained: Request[] = [] + let responseClosed: Promise | undefined + const server = createAdaptorServer({ + fetch: async (req, { outgoing }) => { + counter++ + retained.push(req) + const clone = req.clone() + // eslint-disable-next-line @typescript-eslint/no-explicit-any + registry.register(await (req as any)[bodySharedBufferKey], 'bodyBuffer') + const { data } = (await clone.json()) as { data: string } + responseClosed = new Promise((resolve) => { + outgoing.once('close', () => setTimeout(resolve)) + }) + return new Response(String(data.length)) + }, + }) + + beforeAll( + () => + new Promise((resolve, reject) => { + server.once('error', reject) + server.listen(0, '127.0.0.1', resolve) + }) + ) + + afterAll(() => { + server.close() + }) + + it('Should not have memory leak - cloned and retained request body', async () => { + const res = await requestServer(server, { + method: 'POST', + path: '/', + headers: { 'content-type': 'application/json' }, + body: JSON.stringify({ data: 'x'.repeat(1024) }), + }) + expect(res.status).toBe(200) + expect(await res.text()).toBe('1024') + expect(retained.length).toBe(1) + + // The response has completed and the shared body buffer was released, so + // it must be collectable even though the request is still retained. + await responseClosed + global.gc?.() + await new Promise((resolve) => setTimeout(resolve, 10)) + global.gc?.() + await new Promise((resolve) => setTimeout(resolve, 10)) + expect(counter).toBe(0) + }) +}) + describe('serve', () => { const app = new Hono() app.get('/', (c) => c.newResponse(null, 200))