From 281b4edbb4e565e02db6b33bcaee8982251631fc Mon Sep 17 00:00:00 2001 From: SynthLuvr <131367121+SynthLuvr@users.noreply.github.com> Date: Tue, 25 Aug 2026 12:05:37 +0200 Subject: [PATCH] fix(request): release cloned request bodies via buffer-once replay (#347) clone() no longer tee()s the socket-backed body stream, whose unread branch queues the entire body until it is read or GC'd - pinned for as long as the request graph is reachable (logger/tracing contexts, error objects, profilers). Instead the body is read once into a shared Buffer (via the existing readBodyDirect fast path) and the original request + clones replay from it. When the response completes, the listener releases the shared buffer and cancels replay streams that were never read. - Hook the release of the synchronous cacheable fast path to 'close' rather than 'finish' so the buffer is also released when the client disconnects early, and only attach the listener when a shared body exists - the path stays free of close listeners otherwise - Export bodySharedBufferKey so tests can watch the shared buffer with a FinalizationRegistry, next to the existing memory leak tests --- src/listener.ts | 14 +++++ src/request.ts | 120 +++++++++++++++++++++++++++++++++++++----- test/listener.test.ts | 75 ++++++++++++++++++++++++++ test/request.test.ts | 67 +++++++++++++++++++++++ test/server.test.ts | 65 ++++++++++++++++++++++- 5 files changed, 328 insertions(+), 13 deletions(-) diff --git a/src/listener.ts b/src/listener.ts index c8c27639..b7d15bbc 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 a4b0e110..a2e0c9af 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 14a76901..aa337afd 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 e00e17c4..c7ba36f2 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 387353d4..33c79689 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))