feat: Clean up the incoming object if the request is not completely finished. (#252)

* feat: Clean up the incoming object if the request is not completely finished.

* feat: add autoCleanupIncoming option to disable auto cleanup of incoming request body

* test: add test cases for partially consumed and cancelled request body

* test: added test for incomplete requests with low-level sockets, a pattern that causes automatic cleanup.

* test: ignore test/app.ts in jest.config.js

* test: skip test for partially consumed pattern in v18

* fix: typo

* docs: fix typo

* fix: eliminate environment dependence of type loading

Co-authored-by: Yusuke Wada <[email protected]>

---------

Co-authored-by: Yusuke Wada <[email protected]>

Author: usualoma

README.md

@@ -125,6 +125,18 @@ serve({
 })
 ```
 
+### `autoCleanupIncoming`
+
+The default value is `true`. The Node.js Adapter automatically cleans up (explicitly call `destroy()` method) if application is not finished to consume the incoming request. If you don't want to do that, set `false`.
+If the application accepts connections from arbitrary clients, this cleanup must be done otherwise incomplete requests from clients may cause the application to stop responding. If your application only accepts connections from trusted clients, such as in a reverse proxy environment and there is no process that returns a response without reading the body of the POST request all the way through, you can improve performance by setting it to `false`.
+
+```ts
+serve({
+  fetch: app.fetch,
+  autoCleanupIncoming: false,
+})
+```
+
 ## Middleware
 
 Most built-in middleware also works with Node.js.

jest.config.js

@@ -1,6 +1,6 @@
 module.exports = {
   testMatch: ['**/test/**/*.+(ts)', '**/src/**/(*.)+(test).+(ts)'],
-  modulePathIgnorePatterns: ["test/setup.ts"],
+  modulePathIgnorePatterns: ["test/setup.ts", "test/app.ts"],
   transform: {
     '^.+\\.(ts)$': 'ts-jest',
   },

src/listener.ts

@@ -1,9 +1,12 @@
 import type { IncomingMessage, ServerResponse, OutgoingHttpHeaders } from 'node:http'
-import type { Http2ServerRequest, Http2ServerResponse } from 'node:http2'
+import { Http2ServerRequest } from 'node:http2'
+import type { Http2ServerResponse } from 'node:http2'
+import type { IncomingMessageWithWrapBodyStream } from './request'
 import {
   abortControllerKey,
   newRequest,
   Request as LightweightRequest,
+  wrapBodyStream,
   toRequestError,
 } from './request'
 import { cacheKey, Response as LightweightResponse } from './response'
@@ -13,6 +16,11 @@ import { writeFromReadableStream, buildOutgoingHttpHeaders } from './utils'
 import { X_ALREADY_SENT } from './utils/response/constants'
 import './globals'
 
+const outgoingEnded = Symbol('outgoingEnded')
+type OutgoingHasOutgoingEnded = Http2ServerResponse & {
+  [outgoingEnded]?: () => void
+}
+
 const regBuffer = /^no$/i
 const regContentType = /^(application\/json\b|text\/(?!event-stream\b))/i
 
@@ -78,10 +86,12 @@ const responseViaCache = async (
     outgoing.end(new Uint8Array(await body.arrayBuffer()))
   } else {
     flushHeaders(outgoing)
-    return writeFromReadableStream(body, outgoing)?.catch(
+    await writeFromReadableStream(body, outgoing)?.catch(
       (e) => handleResponseError(e, outgoing) as undefined
     )
   }
+
+  ;(outgoing as OutgoingHasOutgoingEnded)[outgoingEnded]?.()
 }
 
 const responseViaResponseObject = async (
@@ -154,6 +164,8 @@ const responseViaResponseObject = async (
     outgoing.writeHead(res.status, resHeaderRecord)
     outgoing.end()
   }
+
+  ;(outgoing as OutgoingHasOutgoingEnded)[outgoingEnded]?.()
 }
 
 export const getRequestListener = (
@@ -162,8 +174,10 @@ export const getRequestListener = (
     hostname?: string
     errorHandler?: CustomErrorHandler
     overrideGlobalObjects?: boolean
+    autoCleanupIncoming?: boolean
   } = {}
 ) => {
+  const autoCleanupIncoming = options.autoCleanupIncoming ?? true
   if (options.overrideGlobalObjects !== false && global.Request !== LightweightRequest) {
     Object.defineProperty(global, 'Request', {
       value: LightweightRequest,
@@ -185,17 +199,59 @@ export const getRequestListener = (
       // so generate a pseudo Request object with only the minimum required information.
       req = newRequest(incoming, options.hostname)
 
+      let incomingEnded =
+        !autoCleanupIncoming || incoming.method === 'GET' || incoming.method === 'HEAD'
+      if (!incomingEnded) {
+        ;(incoming as IncomingMessageWithWrapBodyStream)[wrapBodyStream] = true
+        incoming.on('end', () => {
+          incomingEnded = true
+        })
+
+        if (incoming instanceof Http2ServerRequest) {
+          // a Http2ServerResponse instance requires additional processing on exit
+          // since outgoing.on('close') is not called even after outgoing.end() is called
+          // when the state is incomplete
+          ;(outgoing as OutgoingHasOutgoingEnded)[outgoingEnded] = () => {
+            // incoming is not consumed to the end
+            if (!incomingEnded) {
+              setTimeout(() => {
+                // in the case of a simple POST request, the cleanup process may be done automatically
+                // and end is called at this point. At that point, nothing is done.
+                if (!incomingEnded) {
+                  setTimeout(() => {
+                    incoming.destroy()
+                    // a Http2ServerResponse instance will not terminate without also calling outgoing.destroy()
+                    outgoing.destroy()
+                  })
+                }
+              })
+            }
+          }
+        }
+      }
+
       // Detect if request was aborted.
       outgoing.on('close', () => {
         const abortController = req[abortControllerKey] as AbortController | undefined
-        if (!abortController) {
-          return
+        if (abortController) {
+          if (incoming.errored) {
+            req[abortControllerKey].abort(incoming.errored.toString())
+          } else if (!outgoing.writableFinished) {
+            req[abortControllerKey].abort('Client connection prematurely closed.')
+          }
         }
 
-        if (incoming.errored) {
-          req[abortControllerKey].abort(incoming.errored.toString())
-        } else if (!outgoing.writableFinished) {
-          req[abortControllerKey].abort('Client connection prematurely closed.')
+        // incoming is not consumed to the end
+        if (!incomingEnded) {
+          setTimeout(() => {
+            // in the case of a simple POST request, the cleanup process may be done automatically
+            // and end is called at this point. At that point, nothing is done.
+            if (!incomingEnded) {
+              setTimeout(() => {
+                incoming.destroy()
+              })
+            }
+          })
         }
       })
 

src/request.ts

@@ -4,6 +4,7 @@
 import type { IncomingMessage } from 'node:http'
 import { Http2ServerRequest } from 'node:http2'
 import { Readable } from 'node:stream'
+import type { ReadableStreamDefaultReader } from 'node:stream/web'
 import type { TLSSocket } from 'node:tls'
 
 export class RequestError extends Error {
@@ -41,6 +42,8 @@ export class Request extends GlobalRequest {
   }
 }
 
+export type IncomingMessageWithWrapBodyStream = IncomingMessage & { [wrapBodyStream]: boolean }
+export const wrapBodyStream = Symbol('wrapBodyStream')
 const newRequestFromIncoming = (
   method: string,
   url: string,
@@ -83,6 +86,23 @@ const newRequestFromIncoming = (
           controller.close()
         },
       })
+    } else if ((incoming as IncomingMessageWithWrapBodyStream)[wrapBodyStream]) {
+      let reader: ReadableStreamDefaultReader<Uint8Array> | undefined
+      init.body = new ReadableStream({
+        async pull(controller) {
+          try {
+            reader ||= Readable.toWeb(incoming).getReader()
+            const { done, value } = await reader.read()
+            if (done) {
+              controller.close()
+            } else {
+              controller.enqueue(value)
+            }
+          } catch (error) {
+            controller.error(error)
+          }
+        },
+      })
     } else {
       // lazy-consume request body
       init.body = Readable.toWeb(incoming) as ReadableStream<Uint8Array>

src/server.ts

@@ -8,6 +8,7 @@ export const createAdaptorServer = (options: Options): ServerType => {
   const requestListener = getRequestListener(fetchCallback, {
     hostname: options.hostname,
     overrideGlobalObjects: options.overrideGlobalObjects,
+    autoCleanupIncoming: options.autoCleanupIncoming,
   })
   // ts will complain about createServerHTTP and createServerHTTP2 not being callable, which works just fine
   // eslint-disable-next-line @typescript-eslint/no-explicit-any

src/types.ts

@@ -70,6 +70,7 @@ export type ServerOptions =
 export type Options = {
   fetch: FetchCallback
   overrideGlobalObjects?: boolean
+  autoCleanupIncoming?: boolean
   port?: number
   hostname?: string
 } & ServerOptions

test/app.ts

@@ -0,0 +1,69 @@
+import { Response as PonyfillResponse } from '@whatwg-node/fetch'
+import { Hono } from 'hono'
+
+export const app = new Hono()
+
+app.get('/', (c) => c.text('Hello! Node!'))
+app.get('/url', (c) => c.text(c.req.url))
+
+app.get('/posts', (c) => {
+  return c.text(`Page ${c.req.query('page')}`)
+})
+app.get('/user-agent', (c) => {
+  return c.text(c.req.header('user-agent') as string)
+})
+app.post('/posts', (c) => {
+  return c.redirect('/posts')
+})
+app.post('/body-consumed', async (c) => {
+  return c.text(`Body length: ${(await c.req.text()).length}`)
+})
+app.post('/no-body-consumed', (c) => {
+  if (!c.req.raw.body) {
+    // force create new request object
+    throw new Error('No body consumed')
+  }
+  return c.text('No body consumed')
+})
+app.post('/body-cancelled', (c) => {
+  if (!c.req.raw.body) {
+    // force create new request object
+    throw new Error('No body consumed')
+  }
+  c.req.raw.body.cancel()
+  return c.text('Body cancelled')
+})
+app.post('/partially-consumed', async (c) => {
+  if (!c.req.raw.body) {
+    // force create new request object
+    throw new Error('No body consumed')
+  }
+  const reader = c.req.raw.body.getReader()
+  await reader.read() // read only one chunk
+  return c.text('Partially consumed')
+})
+app.post('/partially-consumed-and-cancelled', async (c) => {
+  if (!c.req.raw.body) {
+    // force create new request object
+    throw new Error('No body consumed')
+  }
+  const reader = c.req.raw.body.getReader()
+  await reader.read() // read only one chunk
+  reader.cancel()
+  return c.text('Partially consumed and cancelled')
+})
+app.delete('/posts/:id', (c) => {
+  return c.text(`DELETE ${c.req.param('id')}`)
+})
+// @ts-expect-error the response is string
+app.get('/invalid', () => {
+  return '<h1>HTML</h1>'
+})
+app.get('/ponyfill', () => {
+  return new PonyfillResponse('Pony')
+})
+
+app.on('trace', '/', (c) => {
+  const headers = c.req.raw.headers // build new request object
+  return c.text(`headers: ${JSON.stringify(headers)}`)
+})

test/server.test.ts

@@ -1,4 +1,3 @@
-import { Response as PonyfillResponse } from '@whatwg-node/fetch'
 import { Hono } from 'hono'
 import { basicAuth } from 'hono/basic-auth'
 import { compress } from 'hono/compress'
@@ -13,37 +12,9 @@ import { GlobalRequest, Request as LightweightRequest, getAbortController } from
 import { GlobalResponse, Response as LightweightResponse } from '../src/response'
 import { createAdaptorServer, serve } from '../src/server'
 import type { HttpBindings } from '../src/types'
+import { app } from './app'
 
 describe('Basic', () => {
-  const app = new Hono()
-  app.get('/', (c) => c.text('Hello! Node!'))
-  app.get('/url', (c) => c.text(c.req.url))
-
-  app.get('/posts', (c) => {
-    return c.text(`Page ${c.req.query('page')}`)
-  })
-  app.get('/user-agent', (c) => {
-    return c.text(c.req.header('user-agent') as string)
-  })
-  app.post('/posts', (c) => {
-    return c.redirect('/posts')
-  })
-  app.delete('/posts/:id', (c) => {
-    return c.text(`DELETE ${c.req.param('id')}`)
-  })
-  // @ts-expect-error the response is string
-  app.get('/invalid', () => {
-    return '<h1>HTML</h1>'
-  })
-  app.get('/ponyfill', () => {
-    return new PonyfillResponse('Pony')
-  })
-
-  app.on('trace', '/', (c) => {
-    const headers = c.req.raw.headers // build new request object
-    return c.text(`headers: ${JSON.stringify(headers)}`)
-  })
-
   const server = createAdaptorServer(app)
 
   it('Should return 200 response - GET /', async () => {
@@ -82,6 +53,60 @@ describe('Basic', () => {
     expect(res.headers['location']).toBe('/posts')
   })
 
+  it('Should return 200 response - POST /no-body-consumed', async () => {
+    const res = await request(server).post('/no-body-consumed').send('')
+    expect(res.status).toBe(200)
+    expect(res.text).toBe('No body consumed')
+  })
+
+  it('Should return 200 response - POST /body-cancelled', async () => {
+    const res = await request(server).post('/body-cancelled').send('')
+    expect(res.status).toBe(200)
+    expect(res.text).toBe('Body cancelled')
+  })
+
+  it('Should return 200 response - POST /partially-consumed', async () => {
+    const buffer = Buffer.alloc(1024 * 10) // large buffer
+    const res = await new Promise<any>((resolve, reject) => {
+      const req = request(server)
+        .post('/partially-consumed')
+        .set('Content-Length', buffer.length.toString())
+
+      req.write(buffer)
+      req.end((err, res) => {
+        if (err) {
+          reject(err)
+        } else {
+          resolve(res)
+        }
+      })
+    })
+
+    expect(res.status).toBe(200)
+    expect(res.text).toBe('Partially consumed')
+  })
+
+  it('Should return 200 response - POST /partially-consumed-and-cancelled', async () => {
+    const buffer = Buffer.alloc(1) // A large buffer will not make the test go far, so keep it small because it won't go far.
+    const res = await new Promise<any>((resolve, reject) => {
+      const req = request(server)
+        .post('/partially-consumed-and-cancelled')
+        .set('Content-Length', buffer.length.toString())
+
+      req.write(buffer)
+      req.end((err, res) => {
+        if (err) {
+          reject(err)
+        } else {
+          resolve(res)
+        }
+      })
+    })
+
+    expect(res.status).toBe(200)
+    expect(res.text).toBe('Partially consumed and cancelled')
+  })
+
   it('Should return 201 response - DELETE /posts/123', async () => {
     const res = await request(server).delete('/posts/123')
     expect(res.status).toBe(200)