fix aborting Streams (#3754)

* fix aborting Streams

* stop memory leak

lib/web/fetch/index.js

@@ -1943,8 +1943,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) {
@@ -2205,10 +2207,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/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()
+          }
+        }
+      })
+    })
+  }
+)