fix: request draining for early 413 responses (#329)

* fix: request draining for early 413 responses

* fix: close HTTP/2 streams immediately in drainIncoming to avoid ReadableStream controller conflict on Node.js v20

* test: assert http2 early 413 closes without reset

* test: cleanup early 413 upload timer

* test: skip flaky early 413 check on windows

Author: usualoma

src/listener.ts

@@ -1,5 +1,5 @@
 import type { IncomingMessage, ServerResponse, OutgoingHttpHeaders } from 'node:http'
-import { Http2ServerRequest } from 'node:http2'
+import { Http2ServerRequest, constants as h2constants } from 'node:http2'
 import type { Http2ServerResponse } from 'node:http2'
 import type { Writable } from 'node:stream'
 import type { IncomingMessageWithWrapBodyStream } from './request'
@@ -22,9 +22,68 @@ import {
 import { X_ALREADY_SENT } from './utils/response/constants'
 
 const outgoingEnded = Symbol('outgoingEnded')
+const incomingDraining = Symbol('incomingDraining')
 type OutgoingHasOutgoingEnded = Http2ServerResponse & {
   [outgoingEnded]?: () => void
 }
+type IncomingHasDrainState = (IncomingMessage | Http2ServerRequest) & {
+  [incomingDraining]?: boolean
+}
+
+const DRAIN_TIMEOUT_MS = 500
+const MAX_DRAIN_BYTES = 64 * 1024 * 1024
+
+const drainIncoming = (incoming: IncomingMessage | Http2ServerRequest): void => {
+  const incomingWithDrainState = incoming as IncomingHasDrainState
+  if (incoming.destroyed || incomingWithDrainState[incomingDraining]) {
+    return
+  }
+  incomingWithDrainState[incomingDraining] = true
+
+  // HTTP/2: streams are multiplexed, so we can close immediately
+  // without risking TCP RST racing the response.
+  if (incoming instanceof Http2ServerRequest) {
+    try {
+      // eslint-disable-next-line @typescript-eslint/no-explicit-any
+      ;(incoming as any).stream?.close?.(h2constants.NGHTTP2_NO_ERROR)
+    } catch {
+      // stream may already be closed
+    }
+    return
+  }
+
+  let bytesRead = 0
+  const cleanup = () => {
+    clearTimeout(timer)
+    incoming.off('data', onData)
+    incoming.off('end', cleanup)
+    incoming.off('error', cleanup)
+  }
+
+  const forceClose = () => {
+    cleanup()
+    const socket = incoming.socket
+    if (socket && !socket.destroyed) {
+      socket.destroySoon()
+    }
+  }
+
+  const timer = setTimeout(forceClose, DRAIN_TIMEOUT_MS)
+  timer.unref?.()
+
+  const onData = (chunk: Buffer) => {
+    bytesRead += chunk.length
+    if (bytesRead > MAX_DRAIN_BYTES) {
+      forceClose()
+    }
+  }
+
+  incoming.on('data', onData)
+  incoming.on('end', cleanup)
+  incoming.on('error', cleanup)
+
+  incoming.resume()
+}
 
 const handleRequestError = (): Response =>
   new Response(null, {
@@ -264,15 +323,21 @@ export const getRequestListener = (
                 // 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()
+                    drainIncoming(incoming)
                   })
                 }
               })
             }
           }
         }
+
+        // Drain incoming as soon as the response is flushed to the OS,
+        // before the socket is closed, to prevent TCP RST racing the response.
+        outgoing.on('finish', () => {
+          if (!incomingEnded) {
+            drainIncoming(incoming)
+          }
+        })
       }
 
       // Detect if request was aborted.
@@ -294,7 +359,7 @@ export const getRequestListener = (
             // and end is called at this point. At that point, nothing is done.
             if (!incomingEnded) {
               setTimeout(() => {
-                incoming.destroy()
+                drainIncoming(incoming)
               })
             }
           })

test/app.ts

@@ -52,6 +52,13 @@ app.post('/partially-consumed-and-cancelled', async (c) => {
   reader.cancel()
   return c.text('Partially consumed and cancelled')
 })
+app.post('/early-413', (c) => {
+  if (!c.req.raw.body) {
+    // force create new request object
+    throw new Error('No body consumed')
+  }
+  return c.text('Payload Too Large', 413)
+})
 app.delete('/posts/:id', (c) => {
   return c.text(`DELETE ${c.req.param('id')}`)
 })