fix: add inbound upgrade timeout config option (#2995)

Similar to other transports, this lets us close the connection if
encryption/muxer negotiation takes too long.  Defaults to 10s.

Author: achingbrain

packages/transport-tcp/package.json

@@ -65,6 +65,7 @@
     "@multiformats/mafmt": "^12.1.6",
     "@multiformats/multiaddr": "^12.3.3",
     "@types/sinon": "^17.0.3",
+    "any-signal": "^4.1.1",
     "p-defer": "^4.0.1",
     "p-event": "^6.0.1",
     "progress-events": "^1.0.1",

packages/transport-tcp/src/constants.ts

@@ -8,3 +8,6 @@ export const CLOSE_TIMEOUT = 500
 
 // Close the socket if there is no activity after this long in ms
 export const SOCKET_TIMEOUT = 2 * 60000 // 2 mins
+
+// Inbound connection upgrades must complete within this many ms
+export const INBOUND_UPGRADE_TIMEOUT = 10_000

packages/transport-tcp/src/index.ts

@@ -78,6 +78,14 @@ export interface TCPOptions {
    * Options passed to every `net.createServer` for every TCP server
    */
   listenOpts?: TCPSocketOptions
+
+  /**
+   * Upgrading an inbound connection must happen within this many ms otherwise
+   * the connection will be closed.
+   *
+   * @default 10_000
+   */
+  inboundUpgradeTimeout?: number
 }
 
 /**

packages/transport-tcp/src/listener.ts

@@ -1,7 +1,8 @@
 import net from 'net'
 import { AlreadyStartedError, InvalidParametersError, NotStartedError, TypedEventEmitter, setMaxListeners } from '@libp2p/interface'
+import { anySignal } from 'any-signal'
 import { pEvent } from 'p-event'
-import { CODE_P2P } from './constants.js'
+import { CODE_P2P, INBOUND_UPGRADE_TIMEOUT } from './constants.js'
 import { toMultiaddrConnection } from './socket-to-conn.js'
 import {
   getMultiaddrs,
@@ -31,6 +32,7 @@ export interface CloseServerOnMaxConnectionsOpts {
 
 interface Context extends TCPCreateListenerOptions {
   upgrader: Upgrader
+  inboundUpgradeTimeout?: number
   socketInactivityTimeout?: number
   socketCloseTimeout?: number
   maxConnections?: number
@@ -74,6 +76,7 @@ export class TCPListener extends TypedEventEmitter<ListenerEvents> implements Li
   private addr: string
   private readonly log: Logger
   private readonly shutdownController: AbortController
+  private readonly inboundUpgradeTimeout: number
 
   constructor (private readonly context: Context) {
     super()
@@ -85,6 +88,7 @@ export class TCPListener extends TypedEventEmitter<ListenerEvents> implements Li
     setMaxListeners(Infinity, this.shutdownController.signal)
 
     this.log = context.logger.forComponent('libp2p:tcp:listener')
+    this.inboundUpgradeTimeout = context.inboundUpgradeTimeout ?? INBOUND_UPGRADE_TIMEOUT
     this.addr = 'unknown'
     this.server = net.createServer(context, this.onSocket.bind(this))
 
@@ -201,8 +205,14 @@ export class TCPListener extends TypedEventEmitter<ListenerEvents> implements Li
     this.log('new inbound connection %s', maConn.remoteAddr)
     this.sockets.add(socket)
 
+    const signal = anySignal([
+      this.shutdownController.signal,
+      AbortSignal.timeout(this.inboundUpgradeTimeout)
+    ])
+    setMaxListeners(Infinity, signal)
+
     this.context.upgrader.upgradeInbound(maConn, {
-      signal: this.shutdownController.signal
+      signal
     })
       .then(() => {
         this.log('inbound connection upgraded %s', maConn.remoteAddr)
@@ -240,6 +250,9 @@ export class TCPListener extends TypedEventEmitter<ListenerEvents> implements Li
         this.sockets.delete(socket)
         maConn.abort(err)
       })
+      .finally(() => {
+        signal.clear()
+      })
   }
 
   getAddrs (): Multiaddr[] {

packages/transport-tcp/src/tcp.ts

@@ -192,6 +192,7 @@ export class TCP implements Transport<TCPDialEvents> {
     return new TCPListener({
       ...(this.opts.listenOpts ?? {}),
       ...options,
+      inboundUpgradeTimeout: this.opts.inboundUpgradeTimeout,
       maxConnections: this.opts.maxConnections,
       backlog: this.opts.backlog,
       closeServerOnMaxConnections: this.opts.closeServerOnMaxConnections,