fix(client): cancel unused connection attempts

index.d.ts

@@ -328,6 +328,8 @@ export namespace buildConnector {
     localAddress?: string | null
     socketPath?: string | null
     httpSocket?: Socket
+    /** Aborted when no queued request still depends on this connection attempt. */
+    signal?: AbortSignal
   }
 
   type Callback = (...args: [error: null, socket: Socket | TLSSocket] | [error: Error, socket: null]) => void

lib/api/api-request.js

@@ -4,10 +4,18 @@ const assert = require('node:assert')
 const Readable = require('./readable')
 const { InvalidArgumentError, RequestAbortedError } = require('../core/errors')
 const util = require('../core/util')
-const { kAbortBody } = require('../core/symbols')
+const { kAbortBody, kRequestSignal } = require('../core/symbols')
 
 function noop () {}
 
+function setAbort (handler, abort) {
+  if (handler.reason !== null) {
+    abort(handler.reason)
+  } else {
+    handler.abort = abort
+  }
+}
+
 class RequestHandler {
   constructor (opts, callback) {
     if (!opts || typeof opts !== 'object') {
@@ -197,7 +205,10 @@ function request (opts, callback) {
   try {
     const handler = new RequestHandler(opts, callback)
 
-    this.dispatch(opts, handler)
+    this.dispatch({
+      ...opts,
+      [kRequestSignal]: (abort) => setAbort(handler, abort)
+    }, handler)
   } catch (err) {
     if (typeof callback !== 'function') {
       throw err

lib/core/connect.js

@@ -3,7 +3,7 @@
 const net = require('node:net')
 const assert = require('node:assert')
 const util = require('./util')
-const { InvalidArgumentError, ConnectTimeoutError } = require('./errors')
+const { InvalidArgumentError, ConnectTimeoutError, RequestAbortedError } = require('./errors')
 const timers = require('../util/timers')
 
 function noop () {}
@@ -82,7 +82,7 @@ function buildConnector ({ maxCachedSessions, socketPath, timeout, session: cust
   const options = { path: socketPath, ...opts }
   const sessionCache = new SessionCache(maxCachedSessions == null ? 100 : maxCachedSessions)
   timeout = timeout == null ? 10e3 : timeout
-  return function connect ({ hostname, host, protocol, port, servername, localAddress, httpSocket }, callback) {
+  return function connect ({ hostname, host, protocol, port, servername, localAddress, httpSocket, signal }, callback) {
     let socket
     if (protocol === 'https:') {
       if (!tls) {
@@ -153,11 +153,21 @@ function buildConnector ({ maxCachedSessions, socketPath, timeout, session: cust
     }
 
     const clearConnectTimeout = setupConnectTimeout(new WeakRef(socket), { timeout, hostname, port })
+    const removeAbortListener = signal
+      ? util.addAbortListener(signal, () => {
+        util.destroy(socket, signal.reason ?? new RequestAbortedError())
+      })
+      : null
+
+    const cleanup = () => {
+      queueMicrotask(clearConnectTimeout)
+      removeAbortListener?.[Symbol.dispose]()
+    }
 
     socket
       .setNoDelay(true)
       .once(protocol === 'https:' ? 'secureConnect' : 'connect', function () {
-        queueMicrotask(clearConnectTimeout)
+        cleanup()
 
         if (callback) {
           const cb = callback
@@ -166,7 +176,7 @@ function buildConnector ({ maxCachedSessions, socketPath, timeout, session: cust
         }
       })
       .on('error', function (err) {
-        queueMicrotask(clearConnectTimeout)
+        cleanup()
 
         if (callback) {
           const cb = callback

lib/core/request.js

@@ -20,7 +20,8 @@ const {
   assertRequestHandler,
   getServerName,
   normalizedMethodRecords,
-  parseHeaders
+  parseHeaders,
+  addAbortListener
 } = require('./util')
 const { headerNameLowerCasedRecord } = require('./constants')
 
@@ -123,6 +124,7 @@ class Request {
     this.method = method
 
     this.abort = null
+    this.removeEarlyAbortListener = null
     this.blob = null
     this.bodyLength = 0
 
@@ -277,6 +279,9 @@ class Request {
     assert(!this.aborted)
     assert(!this.completed)
 
+    this.removeEarlyAbortListener?.[Symbol.dispose]()
+    this.removeEarlyAbortListener = null
+
     if (this.error) {
       abort(this.error)
     } else {
@@ -343,6 +348,9 @@ class Request {
   }
 
   onFinally () {
+    this.removeEarlyAbortListener?.[Symbol.dispose]()
+    this.removeEarlyAbortListener = null
+
     if (this.errorHandler) {
       this.body.off('error', this.errorHandler)
       this.errorHandler = null
@@ -354,6 +362,10 @@ class Request {
     }
   }
 
+  addEarlyAbortListener (signal, listener) {
+    this.removeEarlyAbortListener = addAbortListener(signal, listener)
+  }
+
   addHeader (key, value) {
     processHeader(this, key, value)
     return this

lib/core/symbols.js

@@ -52,6 +52,7 @@ module.exports = {
   kMaxRequests: Symbol('maxRequestsPerClient'),
   kCounter: Symbol('socket request counter'),
   kMaxResponseSize: Symbol('max response size'),
+  kRequestSignal: Symbol('request signal'),
   kListeners: Symbol('listeners'),
   kHTTPContext: Symbol('http context')
 }

lib/dispatcher/client.js

@@ -9,7 +9,8 @@ const DispatcherBase = require('./dispatcher-base')
 const {
   InvalidArgumentError,
   InformationalError,
-  ClientDestroyedError
+  ClientDestroyedError,
+  RequestAbortedError
 } = require('../core/errors.js')
 const buildConnector = require('../core/connect.js')
 const {
@@ -49,12 +50,14 @@ const {
   kMaxResponseSize,
   kOnError,
   kHTTPContext,
-  kResume
+  kResume,
+  kRequestSignal
 } = require('../core/symbols.js')
 const connectH1 = require('./client-h1.js')
 const { trackSocket } = require('../core/socket-registry.js')
 
 const kClosedResolve = Symbol('kClosedResolve')
+const cancelledConnect = Symbol('cancelled connect')
 
 const getDefaultNodeMaxHeaderSize = http &&
   http.maxHeaderSize &&
@@ -293,6 +296,36 @@ class Client extends DispatcherBase {
     const bodyLength = request.getBodyLength()
 
     this[kQueue].push(request)
+
+    const abort = (reason) => {
+      if (request.aborted || request.completed) {
+        return
+      }
+
+      reason ??= new RequestAbortedError()
+      util.errorRequest(this, request, reason)
+      queueMicrotask(() => this[kResume]())
+    }
+
+    let signalBound = false
+    if (typeof opts[kRequestSignal] === 'function') {
+      signalBound = true
+      try {
+        opts[kRequestSignal](abort)
+      } catch (err) {
+        abort(err)
+      }
+    }
+
+    const { signal } = opts
+    if (!signalBound && signal && (typeof signal.on === 'function' || typeof signal.addEventListener === 'function')) {
+      if (signal.aborted) {
+        abort(signal.reason)
+      } else {
+        request.addEarlyAbortListener(signal, () => abort(signal.reason))
+      }
+    }
+
     if (this[kResuming]) {
       // Do nothing.
     } else if (bodyLength == null && util.isIterable(request.body)) {
@@ -348,6 +381,8 @@ class Client extends DispatcherBase {
         queueMicrotask(callback)
       }
 
+      cancelConnect(this, err)
+
       this[kResume]()
     })
   }
@@ -380,7 +415,7 @@ function onError (client, err) {
  * @param {Client} client
  * @returns
  */
-async function connect (client) {
+async function connect (client, requestDriven = false) {
   assert(!client[kConnecting])
   assert(!client[kHTTPContext])
 
@@ -397,7 +432,11 @@ async function connect (client) {
     hostname = ip
   }
 
-  client[kConnecting] = true
+  const attempt = {
+    controller: new AbortController(),
+    requestDriven
+  }
+  client[kConnecting] = attempt
 
   try {
     const options = {
@@ -406,10 +445,19 @@ async function connect (client) {
       protocol,
       port,
       servername: client[kServerName],
-      localAddress: client[kLocalAddress]
+      localAddress: client[kLocalAddress],
+      signal: attempt.controller.signal
     }
     const socket = await new Promise((resolve, reject) => {
       client[kConnector](options, (err, socket) => {
+        if (client[kConnecting] !== attempt) {
+          if (socket) {
+            util.destroy(socket.on('error', noop), new RequestAbortedError())
+          }
+          resolve(cancelledConnect)
+          return
+        }
+
         if (err) {
           reject(err)
         } else {
@@ -418,20 +466,39 @@ async function connect (client) {
       })
     })
 
+    if (socket === cancelledConnect) {
+      return
+    }
+
+    if (client[kConnecting] !== attempt) {
+      if (socket) {
+        util.destroy(socket.on('error', noop), attempt.controller.signal.reason ?? new RequestAbortedError())
+      }
+      return
+    }
+
     if (client.destroyed) {
       util.destroy(socket.on('error', noop), new ClientDestroyedError())
       return
     }
 
     assert(socket)
 
+    let httpContext
     try {
-      client[kHTTPContext] = await connectH1(client, socket)
+      httpContext = await connectH1(client, socket)
     } catch (err) {
       socket.destroy().on('error', noop)
       throw err
     }
 
+    if (client[kConnecting] !== attempt) {
+      httpContext.destroy(attempt.controller.signal.reason ?? new RequestAbortedError(), noop)
+      return
+    }
+
+    client[kHTTPContext] = httpContext
+
     client[kConnecting] = false
 
     socket[kCounter] = 0
@@ -443,7 +510,7 @@ async function connect (client) {
 
     client.emit('connect', client[kUrl], [client])
   } catch (err) {
-    if (client.destroyed) {
+    if (client.destroyed || client[kConnecting] !== attempt) {
       return
     }
 
@@ -476,6 +543,16 @@ async function connect (client) {
   client[kResume]()
 }
 
+function cancelConnect (client, reason) {
+  const attempt = client[kConnecting]
+  if (!attempt) {
+    return
+  }
+
+  client[kConnecting] = false
+  attempt.controller.abort(reason ?? new RequestAbortedError())
+}
+
 function emitDrain (client) {
   client[kNeedDrain] = 0
   client.emit('drain', client[kUrl], [client])
@@ -505,6 +582,10 @@ function _resume (client, sync) {
       return
     }
 
+    if (client[kConnecting]?.requestDriven && client[kPending] === 0) {
+      cancelConnect(client)
+    }
+
     if (client[kClosedResolve] && !client[kSize]) {
       client[kClosedResolve]()
       client[kClosedResolve] = null
@@ -537,6 +618,11 @@ function _resume (client, sync) {
 
     const request = client[kQueue][client[kPendingIdx]]
 
+    if (request.aborted) {
+      client[kQueue].splice(client[kPendingIdx], 1)
+      continue
+    }
+
     if (client[kUrl].protocol === 'https:' && client[kServerName] !== request.servername) {
       if (client[kRunning] > 0) {
         return
@@ -555,7 +641,7 @@ function _resume (client, sync) {
 
     if (!client[kHTTPContext]) {
       client[kServerName] = request.servername
-      connect(client)
+      connect(client, true)
       return
     }
 
@@ -567,7 +653,7 @@ function _resume (client, sync) {
       return
     }
 
-    if (!request.aborted && client[kHTTPContext].write(request)) {
+    if (client[kHTTPContext].write(request)) {
       client[kPendingIdx]++
     } else {
       client[kQueue].splice(client[kPendingIdx], 1)