handle body errors (#3632)

Co-authored-by: Aras Abbasi <[email protected]>

Author: KhafraDev

lib/web/fetch/index.js

@@ -65,8 +65,6 @@ const { webidl } = require('./webidl')
 const { STATUS_CODES } = require('node:http')
 const GET_OR_HEAD = ['GET', 'HEAD']
 
-const noop = () => {}
-
 const defaultUserAgent = typeof __UNDICI_IS_NODE__ !== 'undefined' || typeof esbuildDetection !== 'undefined'
   ? 'node'
   : 'undici'
@@ -2145,23 +2143,35 @@ async function httpNetworkFetch (
                   finishFlush: zlib.constants.Z_SYNC_FLUSH
                 }))
               } else if (coding === 'deflate') {
-                decoders.push(createInflate())
+                decoders.push(createInflate({
+                  flush: zlib.constants.Z_SYNC_FLUSH,
+                  finishFlush: zlib.constants.Z_SYNC_FLUSH
+                }))
               } else if (coding === 'br') {
-                decoders.push(zlib.createBrotliDecompress())
+                decoders.push(zlib.createBrotliDecompress({
+                  flush: zlib.constants.BROTLI_OPERATION_FLUSH,
+                  finishFlush: zlib.constants.BROTLI_OPERATION_FLUSH
+                }))
               } else {
                 decoders.length = 0
                 break
               }
             }
           }
 
+          const onError = this.onError.bind(this)
+
           resolve({
             status,
             statusText,
             headersList,
             body: decoders.length
-              ? pipeline(this.body, ...decoders, noop)
-              : this.body.on('error', noop)
+              ? pipeline(this.body, ...decoders, (err) => {
+                if (err) {
+                  this.onError(err)
+                }
+              }).on('error', onError)
+              : this.body.on('error', onError)
           })
 
           return true

lib/web/fetch/util.js

@@ -1338,15 +1338,23 @@ function buildContentRange (rangeStart, rangeEnd, fullLength) {
 // interpreted as a zlib stream, otherwise it's interpreted as a
 // raw deflate stream.
 class InflateStream extends Transform {
+  #zlibOptions
+
+  /** @param {zlib.ZlibOptions} [zlibOptions] */
+  constructor (zlibOptions) {
+    super()
+    this.#zlibOptions = zlibOptions
+  }
+
   _transform (chunk, encoding, callback) {
     if (!this._inflateStream) {
       if (chunk.length === 0) {
         callback()
         return
       }
       this._inflateStream = (chunk[0] & 0x0F) === 0x08
-        ? zlib.createInflate()
-        : zlib.createInflateRaw()
+        ? zlib.createInflate(this.#zlibOptions)
+        : zlib.createInflateRaw(this.#zlibOptions)
 
       this._inflateStream.on('data', this.push.bind(this))
       this._inflateStream.on('end', () => this.push(null))
@@ -1365,8 +1373,12 @@ class InflateStream extends Transform {
   }
 }
 
-function createInflate () {
-  return new InflateStream()
+/**
+ * @param {zlib.ZlibOptions} [zlibOptions]
+ * @returns {InflateStream}
+ */
+function createInflate (zlibOptions) {
+  return new InflateStream(zlibOptions)
 }
 
 /**

test/fetch/issue-3616.js

@@ -0,0 +1,48 @@
+'use strict'
+
+const { createServer } = require('node:http')
+const { tspl } = require('@matteo.collina/tspl')
+const { describe, test, after } = require('node:test')
+const { fetch } = require('../..')
+const { once } = require('node:events')
+
+describe('https://github.com/nodejs/undici/issues/3616', () => {
+  const cases = [
+    'x-gzip',
+    'gzip',
+    'deflate',
+    'br'
+  ]
+
+  for (const encoding of cases) {
+    test(encoding, async t => {
+      t = tspl(t, { plan: 2 })
+      const server = createServer((req, res) => {
+        res.writeHead(200, {
+          'Content-Length': '0',
+          Connection: 'close',
+          'Content-Encoding': encoding
+        })
+        res.end()
+      })
+
+      after(() => {
+        server.close()
+      })
+
+      server.listen(0)
+
+      await once(server, 'listening')
+      const result = await fetch(`http://localhost:${server.address().port}/`)
+
+      t.ok(result.body.getReader())
+
+      process.on('uncaughtException', (reason) => {
+        t.fail('Uncaught Exception:', reason, encoding)
+      })
+
+      await new Promise(resolve => setTimeout(resolve, 100))
+      t.ok(true)
+    })
+  }
+})