fix(node): prevent uncatchable crash when canceling a request body stream

packages/node/src/body.test.ts

@@ -10,8 +10,10 @@ import { isAsyncIteratorObject } from '@standardserver/shared'
 import request from 'supertest'
 import { toNodeHttpBody, toStandardBody } from './body'
 import * as EventStreamModule from './event-stream'
+import * as UtilsModule from './utils'
 
 const toEventStreamSpy = vi.spyOn(EventStreamModule, 'toEventStream')
+const toWebReadableStreamSpy = vi.spyOn(UtilsModule, 'toWebReadableStream')
 const generateContentDispositionSpy = vi.spyOn(StandardServerModule, 'generateContentDisposition')
 const getFilenameFromContentDispositionSpy = vi.spyOn(StandardServerModule, 'getFilenameFromContentDisposition')
 
@@ -309,6 +311,8 @@ describe('toStandardBody', () => {
 
       expect(result).toBeInstanceOf(ReadableStream)
       expect(streamedBytes).toEqual(new Uint8Array(body))
+      expect(toWebReadableStreamSpy).toHaveBeenCalledTimes(1)
+      expect(result).toBe(toWebReadableStreamSpy.mock.results[0]!.value)
     })
   })
 
@@ -342,6 +346,8 @@ describe('toStandardBody', () => {
         .send('hello')
 
       expect(standardBody).toBeInstanceOf(ReadableStream)
+      expect(toWebReadableStreamSpy).toHaveBeenCalledTimes(1)
+      expect(standardBody).toBe(toWebReadableStreamSpy.mock.results[0]!.value)
       const reader = (standardBody as ReadableStream).pipeThrough(new TextDecoderStream()).getReader()
       expect(await reader.read()).toEqual({ done: false, value: 'hello' })
     })

packages/node/src/body.ts

@@ -6,6 +6,7 @@ import { Readable } from 'node:stream'
 import { generateContentDisposition, getFilenameFromContentDisposition, resolveStandardBodyHint } from '@standardserver/core'
 import { isAsyncIteratorObject, parseEmptyableJSON, stringifyJSON } from '@standardserver/shared'
 import { toAsyncIteratorObject, toEventStream } from './event-stream'
+import { toWebReadableStream } from './utils'
 
 export interface ToStandardBodyOptions {
   /**
@@ -71,7 +72,7 @@ export async function toStandardBody(
     return _streamToFile(req, fileName ?? 'blob', contentType ?? '')
   }
 
-  return Readable.toWeb(req as Readable)
+  return toWebReadableStream(req)
 }
 
 export interface ToNodeHttpBodyOptions {

packages/node/src/event-stream.test.ts

@@ -2,9 +2,11 @@ import { Readable } from 'node:stream'
 import * as FetchAdapter from '@standardserver/fetch'
 import { isAsyncIteratorObject } from '@standardserver/shared'
 import { toAsyncIteratorObject, toEventStream } from './event-stream'
+import * as UtilsModule from './utils'
 
 const toAsyncIteratorObjectFetch = vi.spyOn(FetchAdapter, 'toAsyncIteratorObject')
 const toEventStreamFetch = vi.spyOn(FetchAdapter, 'toEventStream')
+const toWebReadableStreamSpy = vi.spyOn(UtilsModule, 'toWebReadableStream')
 
 beforeEach(() => {
   vi.clearAllMocks()
@@ -28,7 +30,9 @@ it('toAsyncIteratorObject', async () => {
   expect(await generator.next()).toEqual({ done: false, value: 3 })
   expect(await generator.next()).toEqual({ done: true, value: undefined })
 
+  expect(toWebReadableStreamSpy).toBeCalledTimes(1)
   expect(toAsyncIteratorObjectFetch).toBeCalledTimes(1)
+  expect(toAsyncIteratorObjectFetch).toHaveBeenCalledWith(toWebReadableStreamSpy.mock.results[0]!.value)
 })
 
 it('toEventStream', async () => {

packages/node/src/event-stream.ts

@@ -7,11 +7,12 @@ import {
   toAsyncIteratorObject as toAsyncIteratorObjectFetch,
   toEventStream as toEventStreamFetch,
 } from '@standardserver/fetch'
+import { toWebReadableStream } from './utils'
 
 export function toAsyncIteratorObject(
   stream: Readable,
 ): AsyncIteratorClass<unknown> {
-  return toAsyncIteratorObjectFetch(Readable.toWeb(stream))
+  return toAsyncIteratorObjectFetch(toWebReadableStream(stream))
 }
 
 export interface ToEventStreamOptions extends ToEventStreamOptionsFetch {}

packages/node/src/utils.test.ts

@@ -1,7 +1,13 @@
-import http from 'node:http'
-import http2 from 'node:http2'
-import { connect } from 'node:net'
-import { canWriteToNodeResponse, getNodeResponseError } from './utils'
+import type { AddressInfo } from 'node:net'
+import { Buffer } from 'node:buffer'
+import { appendFile, mkdtemp, rm } from 'node:fs/promises'
+import http, { createServer } from 'node:http'
+import http2, { createServer as createHttp2Server, connect as http2Connect } from 'node:http2'
+import net, { connect } from 'node:net'
+import { tmpdir } from 'node:os'
+import path from 'node:path'
+import { Readable } from 'node:stream'
+import { canWriteToNodeResponse, getNodeResponseError, toWebReadableStream } from './utils'
 
 describe('canWriteToNodeResponse', () => {
   it('on http1 response aborted by client', async ({ onTestFinished }) => {
@@ -312,3 +318,237 @@ describe('getNodeResponseError', () => {
     await handled
   })
 })
+
+describe('toWebReadableStream', () => {
+  /**
+   * Below the 256 KiB flood chunk, so the consumer cancels on its first read
+   * while the request is still streaming — the condition that crashes a bare
+   * `Readable.toWeb`.
+   */
+  const LIMIT = 64 * 1024
+
+  /**
+   * Runs `fn` while recording `uncaughtException`/`unhandledRejection` into the
+   * returned array (also passed to `fn`), so an uncatchable adapter crash lands
+   * there instead of failing the worker. Restores the runner's listeners after.
+   */
+  async function recordUncaught(fn: (crashes: Error[]) => Promise<void>): Promise<Error[]> {
+    const crashes: Error[] = []
+    const record = (err: unknown): void => {
+      crashes.push(err as Error)
+    }
+
+    const prevExceptions = process.listeners('uncaughtException')
+    const prevRejections = process.listeners('unhandledRejection')
+    process.removeAllListeners('uncaughtException')
+    process.removeAllListeners('unhandledRejection')
+    process.on('uncaughtException', record)
+    process.on('unhandledRejection', record)
+
+    try {
+      await fn(crashes)
+      // Let any queued 'data' event reach a (possibly closed) controller.
+      await new Promise(resolve => setTimeout(resolve, 50))
+      return crashes
+    }
+    finally {
+      process.off('uncaughtException', record)
+      process.off('unhandledRejection', record)
+      prevExceptions.forEach(listener => process.on('uncaughtException', listener))
+      prevRejections.forEach(listener => process.on('unhandledRejection', listener))
+    }
+  }
+
+  /**
+   * Consumes a body like an upload handler: spools each chunk to disk (an async
+   * gap that lets more data queue up) and rejects past `LIMIT`, cancelling the
+   * stream while bytes are still arriving.
+   */
+  async function spoolUntilRejected(body: ReadableStream<Uint8Array>, tmpDir: string): Promise<void> {
+    const sink = path.join(await mkdtemp(path.join(tmpDir, 'chunk-')), 'sink')
+    let total = 0
+    try {
+      for await (const chunk of body) {
+        total += chunk.byteLength
+        if (total > LIMIT) {
+          throw new Error('PAYLOAD_TOO_LARGE')
+        }
+        await appendFile(sink, chunk)
+      }
+    }
+    catch {
+      // Mirrors the plugin turning the oversized body into a 413.
+    }
+  }
+
+  /** Streams an oversized HTTP/1 upload, then destroys the socket mid-flight. */
+  function floodAndAbortHttp1(port: number): Promise<void> {
+    return new Promise((resolve) => {
+      const socket = net.connect(port, '127.0.0.1', () => {
+        socket.write('POST / HTTP/1.1\r\nHost: x\r\nContent-Length: 104857600\r\n\r\n')
+        const blob = Buffer.alloc(256 * 1024, 0x61)
+        const interval = setInterval(() => {
+          if (socket.destroyed || !socket.writable) {
+            clearInterval(interval)
+            return
+          }
+          socket.write(blob)
+        }, 0)
+        socket.on('close', () => clearInterval(interval))
+      })
+      socket.on('error', () => {})
+      socket.on('close', () => resolve())
+      setTimeout(() => socket.destroy(), 15)
+    })
+  }
+
+  /** Streams an oversized HTTP/2 upload, then destroys the request mid-flight. */
+  function floodAndAbortHttp2(port: number): Promise<void> {
+    return new Promise((resolve) => {
+      const client = http2Connect(`http://127.0.0.1:${port}`)
+      client.on('error', () => {})
+      const request = client.request({ ':method': 'POST', ':path': '/', 'content-length': '104857600' })
+      const blob = Buffer.alloc(256 * 1024, 0x61)
+      const interval = setInterval(() => {
+        if (request.destroyed || request.closed) {
+          clearInterval(interval)
+          return
+        }
+        request.write(blob)
+      }, 0)
+      request.on('error', () => clearInterval(interval))
+
+      let settled = false
+      const settle = (): void => {
+        if (settled) {
+          return
+        }
+        settled = true
+        clearInterval(interval)
+        client.destroy()
+        resolve()
+      }
+      request.on('close', settle)
+      setTimeout(() => {
+        request.destroy()
+        settle()
+      }, 15)
+    })
+  }
+
+  /**
+   * Boots a server that spools each request body via `wrap`, fires `iterations`
+   * aborted uploads at it, and reports how many completed plus any crashes.
+   * `stopOnCrash` ends the flood at the first crash.
+   */
+  async function runUploadServer(
+    kind: 'http1' | 'http2',
+    wrap: (req: Readable) => ReadableStream<Uint8Array>,
+    iterations: number,
+    stopOnCrash = false,
+  ): Promise<{ handled: number, crashes: Error[] }> {
+    const tmpDir = await mkdtemp(path.join(tmpdir(), `toweb-${kind}-`))
+    let handled = 0
+
+    const listener = async (req: any, res: any): Promise<void> => {
+      await spoolUntilRejected(wrap(req), tmpDir)
+      handled++
+      try {
+        if (!res.headersSent) {
+          res.statusCode = 413
+          res.end('too large')
+        }
+      }
+      catch {
+        // The request stream may already be torn down; the response is best effort.
+      }
+    }
+
+    const server = kind === 'http1' ? createServer(listener) : createHttp2Server(listener)
+    const flood = kind === 'http1' ? floodAndAbortHttp1 : floodAndAbortHttp2
+
+    const crashes = await recordUncaught(async (crashes) => {
+      await new Promise<void>(resolve => server.listen(0, resolve))
+      const { port } = server.address() as AddressInfo
+      for (let i = 0; i < iterations; i++) {
+        if (stopOnCrash && crashes.length) {
+          break
+        }
+        await flood(port)
+        await new Promise(resolve => setTimeout(resolve, 5))
+      }
+    })
+
+    await new Promise<void>(resolve => server.close(() => resolve()))
+    await rm(tmpDir, { recursive: true, force: true })
+    return { handled, crashes }
+  }
+
+  const bare = (req: Readable): ReadableStream<Uint8Array> => Readable.toWeb(req) as ReadableStream<Uint8Array>
+  const wrapped = (req: Readable): ReadableStream<Uint8Array> => toWebReadableStream(req)
+
+  it('converts a raw buffer stream and preserves its bytes as copies', async () => {
+    const chunks = [Buffer.from('hello '), Buffer.from('world'), Buffer.alloc(1024, 7)]
+    const source = Readable.from(chunks)
+
+    const received: Uint8Array[] = []
+    for await (const chunk of toWebReadableStream(source)) {
+      expect(chunk).toBeInstanceOf(Uint8Array)
+      expect(Buffer.isBuffer(chunk)).toBe(false) // copied out of the Node Buffer
+      received.push(chunk)
+    }
+
+    expect(Buffer.concat(received).equals(Buffer.concat(chunks))).toBe(true)
+  })
+
+  it('does not throw when a raw buffer stream is cancelled mid-read', async () => {
+    const source = Readable.from((async function* () {
+      for (let i = 0; i < 10_000; i++) {
+        yield Buffer.alloc(64 * 1024, 0x61)
+      }
+    })())
+
+    let read = 0
+    const crashes = await recordUncaught(async () => {
+      for await (const chunk of toWebReadableStream(source)) {
+        read += chunk.byteLength
+        if (read >= 256 * 1024) {
+          break // cancels the web stream while the source still has data
+        }
+      }
+      await new Promise(resolve => setImmediate(resolve))
+    })
+
+    expect(crashes).toEqual([])
+    expect(read).toBeGreaterThanOrEqual(256 * 1024)
+    expect(source.destroyed).toBe(true) // cancellation still tears the source down
+  })
+
+  it('lets a bare Readable.toWeb crash an aborted HTTP/1 upload (documents the bug)', async () => {
+    const { crashes } = await runUploadServer('http1', bare, 25, true)
+
+    expect(crashes.length).toBeGreaterThan(0)
+    expect(crashes[0]).toMatchObject({ code: 'ERR_INVALID_STATE' })
+  }, 30_000)
+
+  it('keeps an aborted HTTP/1 upload from crashing the process', async () => {
+    const { handled, crashes } = await runUploadServer('http1', wrapped, 25)
+
+    expect(crashes).toEqual([])
+    expect(handled).toBe(25)
+  }, 30_000)
+
+  it('lets a bare Readable.toWeb crash an aborted HTTP/2 upload (documents the bug)', async () => {
+    const { crashes } = await runUploadServer('http2', bare, 25, true)
+
+    expect(crashes.length).toBeGreaterThan(0)
+    expect(crashes[0]).toMatchObject({ code: 'ERR_INVALID_STATE' })
+  }, 30_000)
+
+  it('keeps an aborted HTTP/2 upload from crashing the process', async () => {
+    const { handled, crashes } = await runUploadServer('http2', wrapped, 25)
+
+    expect(crashes).toEqual([])
+    expect(handled).toBe(25)
+  }, 30_000)
+})

packages/node/src/utils.ts

@@ -1,5 +1,54 @@
+import type { Readable } from 'node:stream'
 import type Stream from 'node:stream'
 import type { NodeHttpResponse } from './types'
+import { IncomingMessage } from 'node:http'
+
+/**
+ * A cancel-safe alternative to `Readable.toWeb`.
+ *
+ * Node's adapter enqueues from `'data'` events, so a chunk arriving after the
+ * consumer cancels hits a closed controller and crashes the process with an
+ * uncaught `ERR_INVALID_STATE` (nodejs/node#54205) — on some Node releases even
+ * through an intermediate `TransformStream`. Pulling through the stream's async
+ * iterator makes that impossible: chunks are only enqueued inside `pull`, never
+ * after cancel. The per-chunk copy detaches chunks from Node's pooled `Buffer`
+ * memory.
+ *
+ * Cancel destroys the source, except http1 server requests: they share their
+ * socket with the response, so destroying them would kill an in-flight
+ * response. They are abandoned instead — stalled by backpressure and reclaimed
+ * on connection teardown.
+ */
+export function toWebReadableStream(stream: Readable): ReadableStream<Uint8Array<ArrayBuffer>> {
+  const iterator = stream[Symbol.asyncIterator]()
+  let canceled = false
+
+  return new ReadableStream({
+    async pull(controller) {
+      const { done, value } = await iterator.next()
+
+      if (canceled) {
+        return // a chunk in flight while cancel happened; drop it
+      }
+
+      if (done) {
+        controller.close()
+      }
+      else {
+        controller.enqueue(new Uint8Array(value))
+      }
+    },
+    cancel(reason) {
+      canceled = true
+
+      const isHttp1ServerRequest = stream instanceof IncomingMessage && stream.method !== null
+
+      if (!isHttp1ServerRequest) {
+        stream.destroy(reason instanceof Error ? reason : undefined)
+      }
+    },
+  })
+}
 
 /**
  * Check both the response itself and its underlying stream (http2) are still writable.