fix: don't reject with unusable body when full body was received

Author: BlankParticle

src/request.ts

@@ -261,6 +261,40 @@ const readBodyWithFastPath = <T>(
   })
 }
 
+const normalizeAbortError = (
+  request: Record<string | symbol, any>,
+  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<string | symbol, any>): 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<string | symbol, any>): Promise<Buffer>
 
   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<Buffer>((resolve, reject) => {
@@ -319,17 +362,14 @@ const readBodyDirect = (request: Record<string | symbol, any>): Promise<Buffer>
         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 = () => {

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<string> }> => {
+    // Wrapped in an object so `await` does not flatten the inner promise
+    let resolveBodyRead!: (result: { body: Promise<string> }) => void
+    const bodyRead = new Promise<{ body: Promise<string> }>((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<void>((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')],