fix: check for connection status before storing

It's possible that the remote can close the connection very shortly
after it is opened, so check the connection status before adding it
to the list of connections.

Fixes a memory leak whereby the connection is already closed and
then is never removed from the connections list.

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

@@ -40,21 +40,29 @@ export class ConnectionPruner {
     this.peerStore = components.peerStore
     this.events = components.events
     this.log = components.logger.forComponent('libp2p:connection-manager:connection-pruner')
+    this.maybePruneConnections = this.maybePruneConnections.bind(this)
+  }
 
-    // check the max connection limit whenever a peer connects
-    components.events.addEventListener('connection:open', () => {
-      this.maybePruneConnections()
-        .catch(err => {
-          this.log.error(err)
-        })
-    })
+  start (): void {
+    this.events.addEventListener('connection:open', this.maybePruneConnections)
+  }
+
+  stop (): void {
+    this.events.removeEventListener('connection:open', this.maybePruneConnections)
+  }
+
+  maybePruneConnections (): void {
+    this._maybePruneConnections()
+      .catch(err => {
+        this.log.error('error while pruning connections %e', err)
+      })
   }
 
   /**
    * If we have more connections than our maximum, select some excess connections
    * to prune based on peer value
    */
-  async maybePruneConnections (): Promise<void> {
+  private async _maybePruneConnections (): Promise<void> {
     const connections = this.connectionManager.getConnections()
     const numConnections = connections.length
 

packages/libp2p/src/connection-manager/dial-queue.ts

@@ -153,7 +153,7 @@ export class DialQueue {
       })
     })
 
-    if (existingConnection != null) {
+    if (existingConnection?.status === 'open') {
       this.log('already connected to %a', existingConnection.remoteAddr)
       options.onProgress?.(new CustomProgressEvent('dial-queue:already-connected'))
       return existingConnection

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

@@ -1,4 +1,4 @@
-import { InvalidMultiaddrError, InvalidParametersError, InvalidPeerIdError, NotStartedError, start, stop } from '@libp2p/interface'
+import { ConnectionClosedError, InvalidMultiaddrError, InvalidParametersError, InvalidPeerIdError, NotStartedError, start, stop } from '@libp2p/interface'
 import { PeerMap } from '@libp2p/peer-collections'
 import { defaultAddressSort } from '@libp2p/utils/address-sort'
 import { RateLimiter } from '@libp2p/utils/rate-limiter'
@@ -214,8 +214,6 @@ export class DefaultConnectionManager implements ConnectionManager, Startable {
 
     this.onConnect = this.onConnect.bind(this)
     this.onDisconnect = this.onDisconnect.bind(this)
-    this.events.addEventListener('connection:open', this.onConnect)
-    this.events.addEventListener('connection:close', this.onDisconnect)
 
     // allow/deny lists
     this.allow = (init.allow ?? []).map(ma => multiaddr(ma))
@@ -268,10 +266,6 @@ export class DefaultConnectionManager implements ConnectionManager, Startable {
 
   readonly [Symbol.toStringTag] = '@libp2p/connection-manager'
 
-  isStarted (): boolean {
-    return this.started
-  }
-
   /**
    * Starts the Connection Manager. If Metrics are not enabled on libp2p
    * only event loop and connection limits will be monitored.
@@ -288,11 +282,7 @@ export class DefaultConnectionManager implements ConnectionManager, Startable {
 
         for (const conns of this.connections.values()) {
           for (const conn of conns) {
-            if (conn.direction === 'inbound') {
-              metric.inbound++
-            } else {
-              metric.outbound++
-            }
+            metric[conn.direction]++
           }
         }
 
@@ -356,9 +346,13 @@ export class DefaultConnectionManager implements ConnectionManager, Startable {
       }
     })
 
+    this.events.addEventListener('connection:open', this.onConnect)
+    this.events.addEventListener('connection:close', this.onDisconnect)
+
     await start(
       this.dialQueue,
-      this.reconnectQueue
+      this.reconnectQueue,
+      this.connectionPruner
     )
 
     this.started = true
@@ -369,9 +363,13 @@ export class DefaultConnectionManager implements ConnectionManager, Startable {
    * Stops the Connection Manager
    */
   async stop (): Promise<void> {
+    this.events.removeEventListener('connection:open', this.onConnect)
+    this.events.removeEventListener('connection:close', this.onDisconnect)
+
     await stop(
       this.reconnectQueue,
-      this.dialQueue
+      this.dialQueue,
+      this.connectionPruner
     )
 
     // Close all connections we're tracking
@@ -413,17 +411,19 @@ export class DefaultConnectionManager implements ConnectionManager, Startable {
       return
     }
 
-    const peerId = connection.remotePeer
-    const storedConns = this.connections.get(peerId)
-    let isNewPeer = false
-
-    if (storedConns != null) {
-      storedConns.push(connection)
-    } else {
-      isNewPeer = true
-      this.connections.set(peerId, [connection])
+    if (connection.status !== 'open') {
+      // this can happen when the remote closes the connection immediately after
+      // opening
+      return
     }
 
+    const peerId = connection.remotePeer
+    const isNewPeer = !this.connections.has(peerId)
+    const storedConns = this.connections.get(peerId) ?? []
+    storedConns.push(connection)
+
+    this.connections.set(peerId, storedConns)
+
     // only need to store RSA public keys, all other types are embedded in the peer id
     if (peerId.publicKey != null && peerId.type === 'RSA') {
       await this.peerStore.patch(peerId, {
@@ -441,20 +441,21 @@ export class DefaultConnectionManager implements ConnectionManager, Startable {
    */
   onDisconnect (evt: CustomEvent<Connection>): void {
     const { detail: connection } = evt
+    const peerId = connection.remotePeer
+    const peerConns = this.connections.get(peerId) ?? []
 
-    if (!this.started) {
-      // This can happen when we are in the process of shutting down the node
-      return
-    }
+    // remove closed connection
+    const filteredPeerConns = peerConns.filter(conn => conn.id !== connection.id)
 
-    const peerId = connection.remotePeer
-    let storedConn = this.connections.get(peerId)
+    // update peer connections
+    this.connections.set(peerId, filteredPeerConns)
 
-    if (storedConn != null && storedConn.length > 1) {
-      storedConn = storedConn.filter((conn) => conn.id !== connection.id)
-      this.connections.set(peerId, storedConn)
-    } else if (storedConn != null) {
+    if (filteredPeerConns.length === 0) {
+      // trigger disconnect event if no connections remain
+      this.log('onDisconnect remove all connections for peer %p', peerId)
       this.connections.delete(peerId)
+
+      // broadcast disconnect event
       this.events.safeDispatchEvent('peer:disconnect', { detail: connection.remotePeer })
     }
   }
@@ -478,7 +479,7 @@ export class DefaultConnectionManager implements ConnectionManager, Startable {
   }
 
   async openConnection (peerIdOrMultiaddr: PeerId | Multiaddr | Multiaddr[], options: OpenConnectionOptions = {}): Promise<Connection> {
-    if (!this.isStarted()) {
+    if (!this.started) {
       throw new NotStartedError('Not started')
     }
 
@@ -508,10 +509,8 @@ export class DefaultConnectionManager implements ConnectionManager, Startable {
       priority: options.priority ?? DEFAULT_DIAL_PRIORITY
     })
 
-    if (connection.remotePeer.equals(this.peerId)) {
-      const err = new InvalidPeerIdError('Can not dial self')
-      connection.abort(err)
-      throw err
+    if (connection.status !== 'open') {
+      throw new ConnectionClosedError('Remote closed connection during opening')
     }
 
     let peerConnections = this.connections.get(connection.remotePeer)

packages/libp2p/src/connection/index.ts

@@ -161,13 +161,6 @@ export class ConnectionImpl implements Connection {
     }
 
     try {
-      this.log.trace('closing all streams')
-
-      // close all streams gracefully - this can throw if we're not multiplexed
-      await Promise.all(
-        this.streams.map(async s => s.close(options))
-      )
-
       this.log.trace('closing underlying transport')
 
       // close raw connection
@@ -184,18 +177,19 @@ export class ConnectionImpl implements Connection {
   }
 
   abort (err: Error): void {
+    if (this.status === 'closed') {
+      return
+    }
+
     this.log.error('aborting connection to %a due to error', this.remoteAddr, err)
 
     this.status = 'closing'
-    this.streams.forEach(s => { s.abort(err) })
-
-    this.log.error('all streams aborted', this.streams.length)
 
     // Abort raw connection
     this._abort(err)
 
-    this.timeline.close = Date.now()
     this.status = 'closed'
+    this.timeline.close = Date.now()
   }
 }
 

packages/libp2p/src/upgrader.ts

@@ -1,4 +1,4 @@
-import { InvalidMultiaddrError, TooManyInboundProtocolStreamsError, TooManyOutboundProtocolStreamsError, LimitedConnectionError, setMaxListeners } from '@libp2p/interface'
+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'
@@ -304,6 +304,14 @@ export class DefaultUpgrader implements Upgrader {
         remotePeer = remotePeerId
       }
 
+      // this can happen if we dial a multiaddr without a peer id, we only find
+      // out the identity of the remote after the connection is encrypted
+      if (remotePeer.equals(this.components.peerId)) {
+        const err = new InvalidPeerIdError('Can not dial self')
+        maConn.abort(err)
+        throw err
+      }
+
       upgradedConn = encryptedConn
       if (opts?.muxerFactory != null) {
         muxerFactory = opts.muxerFactory
@@ -326,6 +334,8 @@ export class DefaultUpgrader implements Upgrader {
     } catch (err: any) {
       maConn.log.error('failed to upgrade inbound connection %s %a - %e', direction === 'inbound' ? 'from' : 'to', maConn.remoteAddr, err)
       throw err
+    } finally {
+      signal.clear()
     }
 
     await this.shouldBlockConnection(direction === 'inbound' ? 'denyInboundUpgradedConnection' : 'denyOutboundUpgradedConnection', remotePeer, maConn)
@@ -538,22 +548,22 @@ export class DefaultUpgrader implements Upgrader {
     const _timeline = maConn.timeline
     maConn.timeline = new Proxy(_timeline, {
       set: (...args) => {
-        if (connection != null && args[1] === 'close' && args[2] != null && _timeline.close == null) {
+        if (args[1] === 'close' && args[2] != null && _timeline.close == null) {
           // Wait for close to finish before notifying of the closure
           (async () => {
             try {
               if (connection.status === 'open') {
                 await connection.close()
               }
             } catch (err: any) {
-              connection.log.error('error closing connection after timeline close', err)
+              connection.log.error('error closing connection after timeline close %e', err)
             } finally {
               this.events.safeDispatchEvent('connection:close', {
                 detail: connection
               })
             }
           })().catch(err => {
-            connection.log.error('error thrown while dispatching connection:close event', err)
+            connection.log.error('error thrown while dispatching connection:close event %e', err)
           })
         }
 
@@ -578,32 +588,31 @@ export class DefaultUpgrader implements Upgrader {
       limits,
       logger: this.components.logger,
       newStream: newStream ?? errConnectionNotMultiplexed,
-      getStreams: () => { if (muxer != null) { return muxer.streams } else { return [] } },
+      getStreams: () => {
+        return muxer?.streams ?? []
+      },
       close: async (options?: AbortOptions) => {
-        // Ensure remaining streams are closed gracefully
-        if (muxer != null) {
-          connection.log.trace('close muxer')
-          await muxer.close(options)
-        }
+        // ensure remaining streams are closed gracefully
+        await muxer?.close(options)
 
-        connection.log.trace('close maconn')
         // close the underlying transport
         await maConn.close(options)
-        connection.log.trace('closed maconn')
       },
       abort: (err) => {
         maConn.abort(err)
-        // Ensure remaining streams are aborted
-        if (muxer != null) {
-          muxer.abort(err)
-        }
+
+        // ensure remaining streams are aborted
+        muxer?.abort(err)
       }
     })
 
     this.events.safeDispatchEvent('connection:open', {
       detail: connection
     })
 
+    // @ts-expect-error nah
+    connection.__maConnTimeline = _timeline
+
     return connection
   }