fix: remove abort handler on close

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
+          }
         })
       }
     }
@@ -114,10 +117,7 @@ class RequestHandler extends AsyncResource {
     const body = 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)
+      body.on('close', this.removeAbortListener)
     }
 
     this.callback = null
@@ -156,9 +156,6 @@ class RequestHandler extends AsyncResource {
   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 +176,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'))
       })