fix: ensure that the upgrader applies timeouts to incoming dials (#3000)

Refactors transports to ensure that the upgrader applies the configured incoming upgrade timeout instead of the individual transports.

Where a transport performs part of the upgrade themselves they can obtain an abort signal from the upgrader which will be preconfigured with the correct incoming upgrade timeout.

Deprecates incoming upgrade timeout config keys where they exist for transports, and also the `outboundUpgradeTimeout` key for the connection manager since it's covered by the `dialTimeout` key.

Author: achingbrain

packages/integration-tests/package.json

@@ -39,8 +39,8 @@
     "@libp2p/bootstrap": "^11.0.13",
     "@libp2p/circuit-relay-v2": "^3.1.3",
     "@libp2p/crypto": "^5.0.7",
-    "@libp2p/daemon-client": "^9.0.4",
-    "@libp2p/daemon-server": "^8.0.3",
+    "@libp2p/daemon-client": "^9.0.5",
+    "@libp2p/daemon-server": "^8.0.4",
     "@libp2p/dcutr": "^2.0.12",
     "@libp2p/echo": "^2.1.3",
     "@libp2p/fetch": "^3.0.0",

packages/integration-tests/test/connections.spec.ts

@@ -263,7 +263,7 @@ describe('connections', () => {
         })
       ],
       connectionManager: {
-        outboundUpgradeTimeout: 10
+        dialTimeout: 10
       }
     }, {
       transports: [

packages/interface-compliance-tests/package.json

@@ -119,6 +119,7 @@
     "@multiformats/multiaddr-matcher": "^1.6.0",
     "abortable-iterator": "^5.1.0",
     "aegir": "^45.1.1",
+    "any-signal": "^4.1.1",
     "delay": "^6.0.0",
     "it-all": "^3.0.6",
     "it-byte-stream": "^1.1.0",

packages/interface-compliance-tests/src/mocks/upgrader.ts

@@ -1,5 +1,7 @@
+import { setMaxListeners } from '@libp2p/interface'
+import { anySignal } from 'any-signal'
 import { mockConnection } from './connection.js'
-import type { Libp2pEvents, Connection, MultiaddrConnection, TypedEventTarget, Upgrader, UpgraderOptions } from '@libp2p/interface'
+import type { Libp2pEvents, Connection, MultiaddrConnection, TypedEventTarget, Upgrader, UpgraderOptions, ClearableSignal } from '@libp2p/interface'
 import type { Registrar } from '@libp2p/interface-internal'
 
 export interface MockUpgraderInit {
@@ -16,7 +18,7 @@ class MockUpgrader implements Upgrader {
     this.events = init.events
   }
 
-  async upgradeOutbound (multiaddrConnection: MultiaddrConnection, opts: UpgraderOptions = {}): Promise<Connection> {
+  async upgradeOutbound (multiaddrConnection: MultiaddrConnection, opts: UpgraderOptions): Promise<Connection> {
     const connection = mockConnection(multiaddrConnection, {
       direction: 'outbound',
       registrar: this.registrar,
@@ -28,7 +30,7 @@ class MockUpgrader implements Upgrader {
     return connection
   }
 
-  async upgradeInbound (multiaddrConnection: MultiaddrConnection, opts: UpgraderOptions = {}): Promise<void> {
+  async upgradeInbound (multiaddrConnection: MultiaddrConnection, opts: UpgraderOptions): Promise<void> {
     const connection = mockConnection(multiaddrConnection, {
       direction: 'inbound',
       registrar: this.registrar,
@@ -37,6 +39,16 @@ class MockUpgrader implements Upgrader {
 
     this.events?.safeDispatchEvent('connection:open', { detail: connection })
   }
+
+  createInboundAbortSignal (signal?: AbortSignal): ClearableSignal {
+    const output = anySignal([
+      AbortSignal.timeout(10_000),
+      signal
+    ])
+    setMaxListeners(Infinity, output)
+
+    return output
+  }
 }
 
 export function mockUpgrader (init: MockUpgraderInit = {}): Upgrader {

packages/interface/src/index.ts

@@ -768,6 +768,13 @@ export interface TraceOptions {
   trace?: any
 }
 
+/**
+ * A signal that needs to be cleared when no longer in use
+ */
+export interface ClearableSignal extends AbortSignal {
+  clear(): void
+}
+
 /**
  * When a routing operation involves reading values, these options allow
  * controlling where the values are read from. By default libp2p will check

packages/interface/src/transport.ts

@@ -1,6 +1,6 @@
 import type { Connection, ConnectionLimits, MultiaddrConnection } from './connection.js'
 import type { TypedEventTarget } from './event-target.js'
-import type { AbortOptions } from './index.js'
+import type { AbortOptions, ClearableSignal } from './index.js'
 import type { StreamMuxerFactory } from './stream-muxer.js'
 import type { Multiaddr } from '@multiformats/multiaddr'
 import type { ProgressOptions, ProgressEvent } from 'progress-events'
@@ -58,7 +58,7 @@ export interface CreateListenerOptions {
   upgrader: Upgrader
 }
 
-export interface DialTransportOptions<DialEvents extends ProgressEvent = ProgressEvent> extends AbortOptions, ProgressOptions<DialEvents> {
+export interface DialTransportOptions<DialEvents extends ProgressEvent = ProgressEvent> extends Required<AbortOptions>, ProgressOptions<DialEvents> {
   /**
    * The upgrader turns a MultiaddrConnection into a Connection which should be
    * returned by the transport's dial method
@@ -122,7 +122,7 @@ export enum FaultTolerance {
   NO_FATAL
 }
 
-export interface UpgraderOptions<ConnectionUpgradeEvents extends ProgressEvent = ProgressEvent> extends ProgressOptions<ConnectionUpgradeEvents>, AbortOptions {
+export interface UpgraderOptions<ConnectionUpgradeEvents extends ProgressEvent = ProgressEvent> extends ProgressOptions<ConnectionUpgradeEvents>, Required<AbortOptions> {
   skipEncryption?: boolean
   skipProtection?: boolean
   muxerFactory?: StreamMuxerFactory
@@ -149,4 +149,14 @@ export interface Upgrader {
    * notifies other libp2p components about the new connection
    */
   upgradeInbound(maConn: MultiaddrConnection, opts?: UpgraderOptions<InboundConnectionUpgradeEvents>): Promise<void>
+
+  /**
+   * Used by transports that perform part of the upgrade process themselves and
+   * do some async work. This allows configuring inbound upgrade timeouts from a
+   * single location.
+   *
+   * Regular transports should just pass the signal from their shutdown
+   * controller to `upgradeInbound`.
+   */
+  createInboundAbortSignal (signal: AbortSignal): ClearableSignal
 }

packages/libp2p/src/connection-manager/constants.defaults.ts

@@ -1,17 +1,17 @@
 /**
  * @see https://libp2p.github.io/js-libp2p/interfaces/index._internal_.ConnectionManagerConfig.html#dialTimeout
  */
-export const DIAL_TIMEOUT = 5e3
+export const DIAL_TIMEOUT = 10_000
 
 /**
- * @see https://libp2p.github.io/js-libp2p/interfaces/index._internal_.ConnectionManagerConfig.html#upgradeTimeout
+ * @see https://libp2p.github.io/js-libp2p/interfaces/index._internal_.ConnectionManagerConfig.html#inboundUpgradeTimeout
  */
-export const UPGRADE_TIMEOUT = 3e3
+export const INBOUND_UPGRADE_TIMEOUT = 10_000
 
 /**
  * @see https://libp2p.github.io/js-libp2p/interfaces/index._internal_.ConnectionManagerConfig.html#protocolNegotiationTimeout
  */
-export const PROTOCOL_NEGOTIATION_TIMEOUT = 2e3
+export const PROTOCOL_NEGOTIATION_TIMEOUT = 10_000
 
 /**
  * @see https://libp2p.github.io/js-libp2p/interfaces/index._internal_.ConnectionManagerConfig.html#maxPeerAddrsToDial

packages/libp2p/src/connection-manager/index.ts

@@ -59,15 +59,15 @@ export interface ConnectionManagerInit {
    * How long a dial attempt is allowed to take, including DNS resolution
    * of the multiaddr, opening a socket and upgrading it to a Connection.
    *
-   * @default 5000
+   * @default 10_000
    */
   dialTimeout?: number
 
   /**
    * When a new incoming connection is opened, the upgrade process (e.g.
    * protect, encrypt, multiplex etc) must complete within this number of ms.
    *
-   * @default 3000
+   * @default 10_000
    */
   inboundUpgradeTimeout?: number
 
@@ -77,7 +77,7 @@ export interface ConnectionManagerInit {
    *
    * Does not apply if an abort signal is passed to the `.dial` method.
    *
-   * @default 3000
+   * @deprecated This is handled by `dialTimeout`
    */
   outboundUpgradeTimeout?: number
 
@@ -92,16 +92,18 @@ export interface ConnectionManagerInit {
   /**
    * Outbound protocol negotiation must complete within this number of ms.
    *
-   * Does not apply if an abort signal is passed to the `.dial` method.
+   * Does not apply if an abort signal is passed to the `.dial` or
+   * `.dialProtocol` method of the `ConnectionManager` or the `openStream`
+   * method of the `Connection`.
    *
-   * @default 2000
+   * @default 10_000
    */
   outboundStreamProtocolNegotiationTimeout?: number
 
   /**
    * Inbound protocol negotiation must complete within this number of ms
    *
-   * @default 2000
+   * @default 10_000
    */
   inboundStreamProtocolNegotiationTimeout?: number
 

packages/libp2p/src/libp2p.ts

@@ -18,7 +18,7 @@ import { DefaultPeerRouting } from './peer-routing.js'
 import { RandomWalk } from './random-walk.js'
 import { Registrar } from './registrar.js'
 import { DefaultTransportManager } from './transport-manager.js'
-import { DefaultUpgrader } from './upgrader.js'
+import { Upgrader } from './upgrader.js'
 import { userAgent } from './user-agent.js'
 import * as pkg from './version.js'
 import type { Components } from './components.js'
@@ -113,11 +113,10 @@ export class Libp2p<T extends ServiceMap = ServiceMap> extends TypedEventEmitter
     }
 
     // Set up the Upgrader
-    this.components.upgrader = new DefaultUpgrader(this.components, {
+    this.components.upgrader = new Upgrader(this.components, {
       connectionEncrypters: (init.connectionEncrypters ?? []).map((fn, index) => this.configureComponent(`connection-encryption-${index}`, fn(this.components))),
       streamMuxers: (init.streamMuxers ?? []).map((fn, index) => this.configureComponent(`stream-muxers-${index}`, fn(this.components))),
       inboundUpgradeTimeout: init.connectionManager?.inboundUpgradeTimeout,
-      outboundUpgradeTimeout: init.connectionManager?.outboundUpgradeTimeout,
       inboundStreamProtocolNegotiationTimeout: init.connectionManager?.inboundStreamProtocolNegotiationTimeout ?? init.connectionManager?.protocolNegotiationTimeout,
       outboundStreamProtocolNegotiationTimeout: init.connectionManager?.outboundStreamProtocolNegotiationTimeout ?? init.connectionManager?.protocolNegotiationTimeout
     })

packages/libp2p/src/upgrader.ts

@@ -1,12 +1,13 @@
 import { InvalidMultiaddrError, TooManyInboundProtocolStreamsError, TooManyOutboundProtocolStreamsError, LimitedConnectionError, setMaxListeners, InvalidPeerIdError } from '@libp2p/interface'
 import * as mss from '@libp2p/multistream-select'
 import { peerIdFromString } from '@libp2p/peer-id'
+import { anySignal } from 'any-signal'
 import { CustomProgressEvent } from 'progress-events'
 import { createConnection } from './connection/index.js'
-import { PROTOCOL_NEGOTIATION_TIMEOUT, UPGRADE_TIMEOUT } from './connection-manager/constants.js'
+import { PROTOCOL_NEGOTIATION_TIMEOUT, INBOUND_UPGRADE_TIMEOUT } from './connection-manager/constants.js'
 import { ConnectionDeniedError, ConnectionInterceptedError, EncryptionFailedError, MuxerUnavailableError } from './errors.js'
 import { DEFAULT_MAX_INBOUND_STREAMS, DEFAULT_MAX_OUTBOUND_STREAMS } from './registrar.js'
-import type { Libp2pEvents, AbortOptions, ComponentLogger, MultiaddrConnection, Connection, Stream, ConnectionProtector, NewStreamOptions, ConnectionEncrypter, SecuredConnection, ConnectionGater, TypedEventTarget, Metrics, PeerId, PeerStore, StreamMuxer, StreamMuxerFactory, Upgrader, UpgraderOptions, ConnectionLimits, SecureConnectionOptions, CounterGroup } from '@libp2p/interface'
+import type { Libp2pEvents, AbortOptions, ComponentLogger, MultiaddrConnection, Connection, Stream, ConnectionProtector, NewStreamOptions, ConnectionEncrypter, SecuredConnection, ConnectionGater, TypedEventTarget, Metrics, PeerId, PeerStore, StreamMuxer, StreamMuxerFactory, Upgrader as UpgraderInterface, UpgraderOptions, ConnectionLimits, SecureConnectionOptions, CounterGroup, ClearableSignal } from '@libp2p/interface'
 import type { ConnectionManager, Registrar } from '@libp2p/interface-internal'
 
 interface CreateConnectionOptions {
@@ -40,13 +41,6 @@ export interface UpgraderInit {
    */
   inboundUpgradeTimeout?: number
 
-  /**
-   * An amount of ms by which an outbound connection upgrade must complete
-   *
-   * @default 3000
-   */
-  outboundUpgradeTimeout?: number
-
   /**
    * When a new incoming stream is opened on a multiplexed connection, protocol
    * negotiation on that stream must complete within this many ms
@@ -120,12 +114,11 @@ export interface DefaultUpgraderComponents {
 
 type ConnectionDeniedType = keyof Pick<ConnectionGater, 'denyOutboundConnection' | 'denyInboundEncryptedConnection' | 'denyOutboundEncryptedConnection' | 'denyInboundUpgradedConnection' | 'denyOutboundUpgradedConnection'>
 
-export class DefaultUpgrader implements Upgrader {
+export class Upgrader implements UpgraderInterface {
   private readonly components: DefaultUpgraderComponents
   private readonly connectionEncrypters: Map<string, ConnectionEncrypter>
   private readonly streamMuxers: Map<string, StreamMuxerFactory>
   private readonly inboundUpgradeTimeout: number
-  private readonly outboundUpgradeTimeout: number
   private readonly inboundStreamProtocolNegotiationTimeout: number
   private readonly outboundStreamProtocolNegotiationTimeout: number
   private readonly events: TypedEventTarget<Libp2pEvents>
@@ -148,8 +141,7 @@ export class DefaultUpgrader implements Upgrader {
       this.streamMuxers.set(muxer.protocol, muxer)
     })
 
-    this.inboundUpgradeTimeout = init.inboundUpgradeTimeout ?? UPGRADE_TIMEOUT
-    this.outboundUpgradeTimeout = init.outboundUpgradeTimeout ?? UPGRADE_TIMEOUT
+    this.inboundUpgradeTimeout = init.inboundUpgradeTimeout ?? INBOUND_UPGRADE_TIMEOUT
     this.inboundStreamProtocolNegotiationTimeout = init.inboundStreamProtocolNegotiationTimeout ?? PROTOCOL_NEGOTIATION_TIMEOUT
     this.outboundStreamProtocolNegotiationTimeout = init.outboundStreamProtocolNegotiationTimeout ?? PROTOCOL_NEGOTIATION_TIMEOUT
     this.events = components.events
@@ -177,12 +169,25 @@ export class DefaultUpgrader implements Upgrader {
     }
   }
 
+  createInboundAbortSignal (signal: AbortSignal): ClearableSignal {
+    const output = anySignal([
+      AbortSignal.timeout(this.inboundUpgradeTimeout),
+      signal
+    ])
+    setMaxListeners(Infinity, output)
+
+    return output
+  }
+
   /**
    * Upgrades an inbound connection
    */
-  async upgradeInbound (maConn: MultiaddrConnection, opts: UpgraderOptions = {}): Promise<void> {
+  async upgradeInbound (maConn: MultiaddrConnection, opts: UpgraderOptions): Promise<void> {
     let accepted = false
 
+    // always apply upgrade timeout for incoming upgrades
+    const signal = this.createInboundAbortSignal(opts.signal)
+
     try {
       this.metrics.dials?.increment({
         inbound: true
@@ -196,14 +201,19 @@ export class DefaultUpgrader implements Upgrader {
 
       await this.shouldBlockConnection('denyInboundConnection', maConn)
 
-      await this._performUpgrade(maConn, 'inbound', opts)
+      await this._performUpgrade(maConn, 'inbound', {
+        ...opts,
+        signal
+      })
     } catch (err) {
       this.metrics.errors?.increment({
         inbound: true
       })
 
       throw err
     } finally {
+      signal.clear()
+
       if (accepted) {
         this.components.connectionManager.afterUpgradeInbound()
       }
@@ -213,7 +223,7 @@ export class DefaultUpgrader implements Upgrader {
   /**
    * Upgrades an outbound connection
    */
-  async upgradeOutbound (maConn: MultiaddrConnection, opts: UpgraderOptions = {}): Promise<Connection> {
+  async upgradeOutbound (maConn: MultiaddrConnection, opts: UpgraderOptions): Promise<Connection> {
     try {
       this.metrics.dials?.increment({
         outbound: true
@@ -251,14 +261,6 @@ export class DefaultUpgrader implements Upgrader {
     let muxerFactory: StreamMuxerFactory | undefined
     let cryptoProtocol
 
-    if (opts.signal == null) {
-      maConn.log('no abort signal was passed while trying to upgrade connection, falling back to default timeout')
-
-      const upgradeTimeoutSignal = AbortSignal.timeout(direction === 'inbound' ? this.inboundUpgradeTimeout : this.outboundUpgradeTimeout)
-      setMaxListeners(Infinity, upgradeTimeoutSignal)
-      opts.signal = upgradeTimeoutSignal
-    }
-
     this.components.metrics?.trackMultiaddrConnection(maConn)
 
     maConn.log.trace('starting the %s connection upgrade', direction)