fixup

lib/api/abort-signal.js

@@ -10,8 +10,9 @@ function abort (self) {
   } else if (self.controller?.abort) {
     self.controller.abort(self[kSignal]?.reason)
   } else {
-    self.onError(self[kSignal]?.reason ?? new RequestAbortedError())
+    self.reason = self[kSignal]?.reason ?? new RequestAbortedError()
   }
+  removeSignal(self)
 }
 
 function addSignal (self, signal) {

lib/api/api-connect.js

@@ -1,7 +1,8 @@
 'use strict'
 
+const assert = require('node:assert')
 const { AsyncResource } = require('node:async_hooks')
-const { InvalidArgumentError, RequestAbortedError, SocketError } = require('../core/errors')
+const { InvalidArgumentError, SocketError } = require('../core/errors')
 const util = require('../core/util')
 const { addSignal, removeSignal } = require('./abort-signal')
 
@@ -26,17 +27,21 @@ class ConnectHandler extends AsyncResource {
     this.opaque = opaque || null
     this.responseHeaders = responseHeaders || null
     this.callback = callback
-    this.abort = null
+    this.controller = null
 
+    this.reason = null
     addSignal(this, signal)
   }
 
-  onConnect (abort, context) {
-    if (!this.callback) {
-      throw new RequestAbortedError()
+  onConnect (controller, context) {
+    if (this.reason) {
+      controller.abort(this.reason)
+      return
     }
 
-    this.abort = abort
+    assert(this.callback)
+
+    this.controller = controller
     this.context = context
   }
 

lib/api/api-pipeline.js

@@ -88,7 +88,7 @@ class PipelineHandler extends AsyncResource {
     this.opaque = opaque || null
     this.responseHeaders = responseHeaders || null
     this.handler = handler
-    this.abort = null
+    this.controller = null
     this.context = null
     this.onInfo = onInfo || null
 
@@ -114,14 +114,14 @@ class PipelineHandler extends AsyncResource {
         }
       },
       destroy: (err, callback) => {
-        const { body, req, res, ret, abort } = this
+        const { body, req, res, ret, controller } = this
 
         if (!err && !ret._readableState.endEmitted) {
           err = new RequestAbortedError()
         }
 
-        if (abort && err) {
-          abort()
+        if (controller && err) {
+          controller.abort()
         }
 
         util.destroy(body, err)
@@ -141,19 +141,22 @@ class PipelineHandler extends AsyncResource {
 
     this.res = null
 
+    this.reason = null
     addSignal(this, signal)
   }
 
-  onConnect (abort, context) {
+  onConnect (controller, context) {
     const { ret, res } = this
 
-    assert(!res, 'pipeline cannot be retried')
-
-    if (ret.destroyed) {
-      throw new RequestAbortedError()
+    if (this.reason) {
+      controller.abort(this.reason)
+      return
     }
 
-    this.abort = abort
+    assert(!res, 'pipeline cannot be retried')
+    assert(!ret.destroyed)
+
+    this.controller = controller
     this.context = context
   }
 

lib/api/api-request.js

@@ -1,10 +1,8 @@
 'use strict'
 
+const assert = require('node:assert')
 const { Readable } = require('./readable')
-const {
-  InvalidArgumentError,
-  RequestAbortedError
-} = require('../core/errors')
+const { InvalidArgumentError } = require('../core/errors')
 const util = require('../core/util')
 const { getResolveErrorBodyCallback } = require('./util')
 const { AsyncResource } = require('node:async_hooks')
@@ -57,21 +55,26 @@ class RequestHandler extends AsyncResource {
     this.onInfo = onInfo || null
     this.throwOnError = throwOnError
     this.highWaterMark = highWaterMark
+    this.controller = null
 
     if (util.isStream(body)) {
       body.on('error', (err) => {
         this.onError(err)
       })
     }
 
+    this.reason = null
     addSignal(this, signal)
   }
 
   onConnect (controller, context) {
-    if (!this.callback) {
-      throw new RequestAbortedError()
+    if (this.reason) {
+      controller.abort(this.reason)
+      return
     }
 
+    assert(this.callback)
+
     this.controller = controller
     this.context = context
   }

lib/api/api-stream.js

@@ -1,10 +1,10 @@
 'use strict'
 
+const assert = require('node:assert')
 const { finished, PassThrough } = require('node:stream')
 const {
   InvalidArgumentError,
-  InvalidReturnValueError,
-  RequestAbortedError
+  InvalidReturnValueError
 } = require('../core/errors')
 const util = require('../core/util')
 const { getResolveErrorBodyCallback } = require('./util')
@@ -53,7 +53,7 @@ class StreamHandler extends AsyncResource {
     this.factory = factory
     this.callback = callback
     this.res = null
-    this.abort = null
+    this.controller = null
     this.context = null
     this.trailers = null
     this.body = body
@@ -66,15 +66,19 @@ class StreamHandler extends AsyncResource {
       })
     }
 
+    this.reason = null
     addSignal(this, signal)
   }
 
-  onConnect (abort, context) {
-    if (!this.callback) {
-      throw new RequestAbortedError()
+  onConnect (controller, context) {
+    if (this.reason) {
+      controller.abort(this.reason)
+      return
     }
 
-    this.abort = abort
+    assert(this.callback)
+
+    this.controller = controller
     this.context = context
   }
 
@@ -126,7 +130,7 @@ class StreamHandler extends AsyncResource {
 
       // TODO: Avoid finished. It registers an unnecessary amount of listeners.
       finished(res, { readable: false }, (err) => {
-        const { callback, res, opaque, trailers, abort } = this
+        const { callback, res, opaque, trailers, controller } = this
 
         this.res = null
         if (err || !res.readable) {
@@ -137,7 +141,7 @@ class StreamHandler extends AsyncResource {
         this.runInAsyncScope(callback, null, err || null, { opaque, trailers })
 
         if (err) {
-          abort()
+          controller.abort()
         }
       })
     }

lib/api/api-upgrade.js

@@ -1,6 +1,6 @@
 'use strict'
 
-const { InvalidArgumentError, RequestAbortedError, SocketError } = require('../core/errors')
+const { InvalidArgumentError, SocketError } = require('../core/errors')
 const { AsyncResource } = require('node:async_hooks')
 const util = require('../core/util')
 const { addSignal, removeSignal } = require('./abort-signal')
@@ -27,18 +27,22 @@ class UpgradeHandler extends AsyncResource {
     this.responseHeaders = responseHeaders || null
     this.opaque = opaque || null
     this.callback = callback
-    this.abort = null
+    this.controller = null
     this.context = null
 
+    this.reason = null
     addSignal(this, signal)
   }
 
-  onConnect (abort, context) {
-    if (!this.callback) {
-      throw new RequestAbortedError()
+  onConnect (controller, context) {
+    if (this.reason) {
+      controller.abort(this.reason)
+      return
     }
 
-    this.abort = abort
+    assert(this.callback)
+
+    this.controller = controller
     this.context = null
   }
 

lib/core/request.js

@@ -229,9 +229,9 @@ class Request {
       abort.paused = false
       abort.aborted = false
       abort.internal = null
-      abort.abort = function (...args) {
+      abort.abort = function (err) {
         abort.aborted = true
-        abort(...args)
+        abort(err)
       }
       abort.pause = function () {
         abort.paused = true

lib/dispatcher/client-h2.js

@@ -458,7 +458,7 @@ function writeH2 (client, request) {
       stream.pause()
     }
 
-    if (request.onHeaders(Number(statusCode), realHeaders, resume, '') === false) {
+    if (request.onHeaders(Number(statusCode), parseH2Headers(realHeaders), resume, '') === false) {
       stream.pause()
     }
 

lib/mock/mock-utils.js

@@ -293,6 +293,7 @@ function mockDispatch (opts, handler) {
         super()
         this.resume = this.resume.bind(this)
         this.pause = this.pause.bind(this)
+        this.abort = this.abort.bind(this)
       }
 
       resume () {
@@ -303,7 +304,8 @@ function mockDispatch (opts, handler) {
         this.#paused = true
       }
 
-      abort () {
+      abort (err) {
+        handler.onError(err)
         this.#aborted = true
       }
 

test/client-request.js

@@ -14,6 +14,10 @@ const { promisify } = require('node:util')
 const { NotSupportedError } = require('../lib/core/errors')
 const { parseFormDataString } = require('./utils/formdata')
 
+process.on('uncatchedException', (err) => {
+  console.error('###', err)
+})
+
 test('request dump big', async (t) => {
   t = tspl(t, { plan: 3 })
 
@@ -140,7 +144,7 @@ test('request hwm', async (t) => {
 })
 
 test('request abort before headers', async (t) => {
-  t = tspl(t, { plan: 6 })
+  t = tspl(t, { plan: 4 })
 
   const signal = new EE()
   const server = createServer((req, res) => {
@@ -162,7 +166,6 @@ test('request abort before headers', async (t) => {
         t.ok(err instanceof errors.RequestAbortedError)
         t.strictEqual(signal.listenerCount('abort'), 0)
       })
-      t.strictEqual(signal.listenerCount('abort'), 1)
 
       client.request({
         path: '/',
@@ -172,7 +175,6 @@ test('request abort before headers', async (t) => {
         t.ok(err instanceof errors.RequestAbortedError)
         t.strictEqual(signal.listenerCount('abort'), 0)
       })
-      t.strictEqual(signal.listenerCount('abort'), 2)
     })
   })
 

test/node-test/client-dispatch.js

@@ -875,15 +875,15 @@ test('dispatches in expected order', async (t) => {
       method: 'POST',
       body: 'body'
     }, {
-      onConnect () {
+      onConnect (controller) {
+        assert(typeof controller.resume === 'function')
+        assert(typeof controller.pause === 'function')
         dispatches.push('onConnect')
       },
       onBodySent () {
         dispatches.push('onBodySent')
       },
-      onStart (controller) {
-        assert(typeof controller.resume === 'function')
-        assert(typeof controller.pause === 'function')
+      onStart () {
         dispatches.push('onStart')
       },
       onResponseStarted () {
@@ -937,15 +937,15 @@ test('dispatches in expected order for http2', async (t) => {
       method: 'POST',
       body: 'body'
     }, {
-      onConnect () {
+      onConnect (controller) {
+        assert(typeof controller.resume === 'function')
+        assert(typeof controller.pause === 'function')
         dispatches.push('onConnect')
       },
       onBodySent () {
         dispatches.push('onBodySent')
       },
-      onStart (controller) {
-        assert(typeof controller.resume === 'function')
-        assert(typeof controller.pause === 'function')
+      onStart () {
         dispatches.push('onStart')
       },
       onResponseStarted () {

types/dispatcher.d.ts

@@ -219,14 +219,16 @@ declare namespace Dispatcher {
   export interface Controller {
     pause (): void;
     resume (): void;
+    abort (err?:Error): void;
     readonly paused: boolean;
+    readonly aborted: boolean;
   }
 
   export interface DispatchHandlers {
     /** Invoked before request is dispatched on socket. May be invoked multiple times when a request is retried when the request at the head of the pipeline fails. */
-    onConnect?(abort: () => void): void;
+    onConnect?(controller: Controller): void;
     /** */
-    onStart?(controller: Controller): void;
+    onStart?(): void;
     /** Invoked when an error has occurred. */
     onError?(err: Error): void;
     /** Invoked when request is upgraded either due to a `Upgrade` header or `CONNECT` method. */