fix: remove abort handler on close (nodejs#3211)

Author: ronag

lib/api/api-request.js

@@ -70,15 +70,18 @@ class RequestHandler extends AsyncResource {
         this.reason = this.signal.reason ?? new RequestAbortedError()
       } else {
         this.removeAbortListener = util.addAbortListener(this.signal, () => {
-          this.removeAbortListener?.()
-          this.removeAbortListener = null
-
           this.reason = this.signal.reason ?? new RequestAbortedError()
           if (this.res) {
             util.destroy(this.res, this.reason)
           } else if (this.abort) {
             this.abort(this.reason)
           }
+
+          if (this.removeAbortListener) {
+            this.res?.off('close', this.removeAbortListener)
+            this.removeAbortListener()
+            this.removeAbortListener = null
+          }
         })
       }
     }
@@ -111,54 +114,44 @@ class RequestHandler extends AsyncResource {
     const parsedHeaders = responseHeaders === 'raw' ? util.parseHeaders(rawHeaders) : headers
     const contentType = parsedHeaders['content-type']
     const contentLength = parsedHeaders['content-length']
-    const body = new Readable({ resume, abort, contentType, contentLength, highWaterMark })
+    const res = new Readable({ resume, abort, contentType, contentLength, highWaterMark })
 
     if (this.removeAbortListener) {
-      // TODO (fix): 'close' is sufficient but breaks tests.
-      body
-        .on('end', this.removeAbortListener)
-        .on('error', this.removeAbortListener)
+      res.on('close', this.removeAbortListener)
     }
 
     this.callback = null
-    this.res = body
+    this.res = res
     if (callback !== null) {
       if (this.throwOnError && statusCode >= 400) {
         this.runInAsyncScope(getResolveErrorBodyCallback, null,
-          { callback, body, contentType, statusCode, statusMessage, headers }
+          { callback, body: res, contentType, statusCode, statusMessage, headers }
         )
       } else {
         this.runInAsyncScope(callback, null, null, {
           statusCode,
           headers,
           trailers: this.trailers,
           opaque,
-          body,
+          body: res,
           context
         })
       }
     }
   }
 
   onData (chunk) {
-    const { res } = this
-    return res.push(chunk)
+    return this.res.push(chunk)
   }
 
   onComplete (trailers) {
-    const { res } = this
-
     util.parseHeaders(trailers, this.trailers)
-
-    res.push(null)
+    this.res.push(null)
   }
 
   onError (err) {
     const { res, callback, body, opaque } = this
 
-    this.removeAbortListener?.()
-    this.removeAbortListener = null
-
     if (callback) {
       // TODO: Does this need queueMicrotask?
       this.callback = null
@@ -179,6 +172,12 @@ class RequestHandler extends AsyncResource {
       this.body = null
       util.destroy(body, err)
     }
+
+    if (this.removeAbortListener) {
+      res?.off('close', this.removeAbortListener)
+      this.removeAbortListener()
+      this.removeAbortListener = null
+    }
   }
 }
 

test/client.js

@@ -63,7 +63,7 @@ test('basic get', async (t) => {
       body.on('data', (buf) => {
         bufs.push(buf)
       })
-      body.on('end', () => {
+      body.on('close', () => {
         t.strictEqual(signal.listenerCount('abort'), 0)
         t.strictEqual('hello', Buffer.concat(bufs).toString('utf8'))
       })
@@ -135,7 +135,7 @@ test('basic get with custom request.reset=true', async (t) => {
       body.on('data', (buf) => {
         bufs.push(buf)
       })
-      body.on('end', () => {
+      body.on('close', () => {
         t.strictEqual(signal.listenerCount('abort'), 0)
         t.strictEqual('hello', Buffer.concat(bufs).toString('utf8'))
       })