fix(request): release cloned request bodies via buffer-once replay

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

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<never> => {
   return Promise.reject(newBodyUnusableError())
 }
 
+const createBufferReplayStream = (
+  bufferPromise: Promise<Buffer>,
+  streams: Set<ReadableStream<Uint8Array>>
+): ReadableStream<Uint8Array> => {
+  const stream = new ReadableStream<Uint8Array>({
+    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<string | symbol, any>): Promise<Buffer>
   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<string | symbol, any>): Promise<Buffer> => {
+  if (!request[bodySharedBufferKey]) {
+    const raw = readRawBodyIfAvailable(request)
+    request[bodySharedBufferKey] = raw ? Promise.resolve(raw) : readBodyDirect(request)
+  }
+  const bufferPromise = request[bodySharedBufferKey] as Promise<Buffer>
+  bufferPromise.then(
+    (buffer) => {
+      if (!request[bodyReleasedKey]) {
+        request[bodyBufferKey] ||= buffer
+      }
+    },
+    () => {} // readers surface the error via their replay stream
+  )
+  return bufferPromise
+}
+
+const newReplayRequest = (
+  request: Record<string | symbol, any>,
+  method: string,
+  bufferPromise: Promise<Buffer>
+): Request => {
+  const replayStreams = (request[bodySharedStreamsKey] ??= new Set<ReadableStream<Uint8Array>>())
+  return new Request(
+    request[urlKey] as string,
+    {
+      method,
+      headers: request.headers,
+      signal: request[getAbortController]().signal,
+      body: createBufferReplayStream(bufferPromise, replayStreams),
+    } as RequestInit
+  )
+}
+
 const requestPrototype: Record<string | symbol, any> = {
   get method() {
     return this[methodKey]
@@ -580,6 +644,22 @@ const requestPrototype: Record<string | symbol, any> = {
     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<string | symbol, any> = {
       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()

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()
 

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)