fix(node): stop crashing when a web stream's source errors before its first read

packages/node/src/utils.test.ts

@@ -651,6 +651,33 @@ describe('toWebReadableStream', () => {
     expect(source.destroyed).toBe(true) // cancellation still tears the source down
   })
 
+  it('does not throw when cancelled with an error before any read', async () => {
+    const source = new Readable({ read() {} })
+    const reason = new Error('cancelled')
+
+    const crashes = await recordUncaught(() => toWebReadableStream(source).cancel(reason))
+
+    expect(crashes).toEqual([])
+    expect(source.errored).toBe(reason)
+  })
+
+  it('does not throw when the source errors in the tick it is wrapped', async () => {
+    const source = new Readable({ read() {} })
+    const error = new Error('failed')
+
+    const crashes = await recordUncaught(async () => {
+      const body = await new Promise<ReadableStream>((resolve) => {
+        setImmediate(() => {
+          resolve(toWebReadableStream(source))
+          source.destroy(error)
+        })
+      })
+      await expect(body.getReader().read()).rejects.toBe(error)
+    })
+
+    expect(crashes).toEqual([])
+  })
+
   it('keeps an aborted HTTP/1 upload from crashing the process', async () => {
     const { handled, crashes } = await runUploadServer('http1', 25)
 
@@ -754,4 +781,16 @@ describe('toWebReadableStream', () => {
     expect(received).toBe('ok')
     expect(request.rstCode).toBe(http2.constants.NGHTTP2_NO_ERROR)
   })
+
+  it('ignores a request torn down while its cancelled body drains', async () => {
+    const req = new http.IncomingMessage(new net.Socket())
+    req.method = 'POST'
+
+    const crashes = await recordUncaught(async () => {
+      await toWebReadableStream(req).cancel()
+      req.destroy()
+    })
+
+    expect(crashes).toEqual([])
+  })
 })

packages/node/src/utils.ts

@@ -28,6 +28,7 @@ import type { NodeHttpResponse } from './types'
  */
 export function toWebReadableStream(stream: Readable): ReadableStream<Uint8Array<ArrayBuffer>> {
   const iterator = stream[Symbol.asyncIterator]()
+  stream.on('error', () => {})
   let canceled = false
 
   return new ReadableStream({