[v6.x] fix(fetch): fix aborting Streams (#3754) (#6000)

* fix aborting Streams (#3754)

* fix aborting Streams

* stop memory leak

(cherry picked from commit afeb626)

* test: add regression test for invalid compressed fetch body

A response that advertises content-encoding: gzip but delivers a body
that is not valid gzip in a later packet than the headers left the
body stream pending forever, because the 'terminated' listener had
already been removed in onComplete.

Refs: #5999

---------

Co-authored-by: Austin Henrie <113467168+epistemancering@users.noreply.github.com>

Author: mrityunjay-pandey

lib/web/fetch/index.js

@@ -1950,8 +1950,10 @@ async function httpNetworkFetch (
   // 19. Run these steps in parallel:
 
   //    1. Run these steps, but abort when fetchParams is canceled:
-  fetchParams.controller.onAborted = onAborted
-  fetchParams.controller.on('terminated', onAborted)
+  if (!fetchParams.controller.resume) {
+    fetchParams.controller.on('terminated', onAborted)
+  }
+
   fetchParams.controller.resume = async () => {
     // 1. While true
     while (true) {
@@ -2218,10 +2220,6 @@ async function httpNetworkFetch (
             fetchParams.controller.off('terminated', this.abort)
           }
 
-          if (fetchParams.controller.onAborted) {
-            fetchParams.controller.off('terminated', fetchParams.controller.onAborted)
-          }
-
           fetchParams.controller.ended = true
 
           this.body.push(null)

test/fetch/encoding.js

@@ -122,3 +122,27 @@ describe('content-encoding chain limit', () => {
     )
   })
 })
+
+// https://github.com/nodejs/undici/issues/5999
+test('invalid compressed body received after the headers rejects instead of hanging', async (t) => {
+  const server = createServer((req, res) => {
+    res.writeHead(200, {
+      'Content-Encoding': 'gzip',
+      'Content-Length': '16'
+    })
+    res.flushHeaders()
+
+    setTimeout(() => res.end(Buffer.alloc(16)), 100)
+  }).listen(0, '127.0.0.1')
+
+  t.after(closeServerAsPromise(server))
+  await once(server, 'listening')
+
+  const response = await fetch(`http://127.0.0.1:${server.address().port}`, { keepalive: false })
+
+  await assert.rejects(response.text(), (err) => {
+    assert.strictEqual(err.name, 'TypeError')
+    assert.strictEqual(err.cause?.code, 'Z_DATA_ERROR')
+    return true
+  })
+})

test/fetch/issue-1711.js

@@ -31,3 +31,30 @@ test('Redirecting a bunch does not cause a MaxListenersExceededWarning', async (
 
   assert.deepStrictEqual(response.url, `${url}/${redirects - 1}`)
 })
+
+test(
+  'aborting a Stream throws',
+  () => {
+    return new Promise((resolve, reject) => {
+      const httpServer = createServer((request, response) => {
+        response.end(new Uint8Array(20000))
+      }).listen(async () => {
+        const serverAddress = httpServer.address()
+
+        if (typeof serverAddress === 'object') {
+          const abortController = new AbortController()
+          const readStream = (await fetch(`http://localhost:${serverAddress?.port}`, { signal: abortController.signal })).arrayBuffer()
+          abortController.abort()
+          setTimeout(reject)
+
+          try {
+            await readStream
+          } catch {
+            httpServer.close()
+            resolve()
+          }
+        }
+      })
+    })
+  }
+)