fix: opt-in to toplogy notifications on transient connections (#2049)

Adds a `notifyOnTransient` option when registering a network topology to opt-in to being notified when peers that support the registered protocol connect over transient connections.  False by default.

The logic has been switched to check each connection's transient property and only notify if the user has opted in.

The side effect here is that if `notifyOnTransient` is true, and the peer ends up opening a direct connection (for example they dial us via circuit relay, open a stream to do the WebRTC SDP exchange, then open a WebRTC connection), identify will run on the second connection so the topology will receive two notifications.

This is not a breaking change since the previous behaviour would have been to only notify on the initial transient connection, which you can't do data-heavy things like bitswap over, or long-lived things like GossipSub so is probably a bug.

Fixes #2036

---------

Co-authored-by: Chad Nehemiah <[email protected]>

Author: achingbrain

packages/interface/src/index.ts

@@ -99,6 +99,11 @@ export interface IdentifyResult {
    * If sent by the remote peer this is the deserialized signed peer record
    */
   signedPeerRecord?: SignedPeerRecord
+
+  /**
+   * The connection that the identify protocol ran over
+   */
+  connection: Connection
 }
 
 /**

packages/interface/src/topology/index.ts

@@ -5,6 +5,21 @@ export interface Topology {
   min?: number
   max?: number
 
+  /**
+   * If true, invoke `onConnect` for this topology on transient (e.g. short-lived
+   * and/or data-limited) connections. (default: false)
+   */
+  notifyOnTransient?: boolean
+
+  /**
+   * Invoked when a new connection is opened to a peer that supports the
+   * registered protocol
+   */
   onConnect?(peerId: PeerId, conn: Connection): void
+
+  /**
+   * Invoked when the last connection to a peer that supports the registered
+   * protocol closes
+   */
   onDisconnect?(peerId: PeerId): void
 }

packages/libp2p/src/circuit-relay/transport/discovery.ts

@@ -57,6 +57,7 @@ export class RelayDiscovery extends TypedEventEmitter<RelayDiscoveryEvents> impl
     // register a topology listener for when new peers are encountered
     // that support the hop protocol
     this.topologyId = await this.registrar.register(RELAY_V2_HOP_CODEC, {
+      notifyOnTransient: true,
       onConnect: (peerId) => {
         this.safeDispatchEvent('relay:discover', { detail: peerId })
       }

packages/libp2p/src/dcutr/dcutr.ts

@@ -72,6 +72,7 @@ export class DefaultDCUtRService implements Startable {
     // register for notifications of when peers that support DCUtR connect
     // nb. requires the identify service to be enabled
     this.topologyId = await this.registrar.register(multicodec, {
+      notifyOnTransient: true,
       onConnect: (peerId, connection) => {
         if (!connection.transient) {
           // the connection is already direct, no upgrade is required

packages/libp2p/src/identify/identify.ts

@@ -315,22 +315,7 @@ export class DefaultIdentifyService implements Startable, IdentifyService {
       this.addressManager.addObservedAddr(cleanObservedAddr)
     }
 
-    const signedPeerRecord = await this.#consumeIdentifyMessage(connection.remotePeer, message)
-
-    const result: IdentifyResult = {
-      peerId: id,
-      protocolVersion: message.protocolVersion,
-      agentVersion: message.agentVersion,
-      publicKey: message.publicKey,
-      listenAddrs: message.listenAddrs.map(buf => multiaddr(buf)),
-      observedAddr: message.observedAddr == null ? undefined : multiaddr(message.observedAddr),
-      protocols: message.protocols,
-      signedPeerRecord
-    }
-
-    this.events.safeDispatchEvent('peer:identify', { detail: result })
-
-    return result
+    return this.#consumeIdentifyMessage(connection, message)
   }
 
   /**
@@ -405,7 +390,7 @@ export class DefaultIdentifyService implements Startable, IdentifyService {
       const message = await pb.read(options)
       await stream.close(options)
 
-      await this.#consumeIdentifyMessage(connection.remotePeer, message)
+      await this.#consumeIdentifyMessage(connection, message)
     } catch (err: any) {
       log.error('received invalid message', err)
       stream.abort(err)
@@ -415,8 +400,8 @@ export class DefaultIdentifyService implements Startable, IdentifyService {
     log('handled push from %p', connection.remotePeer)
   }
 
-  async #consumeIdentifyMessage (remotePeer: PeerId, message: Identify): Promise<SignedPeerRecord | undefined> {
-    log('received identify from %p', remotePeer)
+  async #consumeIdentifyMessage (connection: Connection, message: Identify): Promise<IdentifyResult> {
+    log('received identify from %p', connection.remotePeer)
 
     if (message == null) {
       throw new Error('Message was null or undefined')
@@ -436,7 +421,7 @@ export class DefaultIdentifyService implements Startable, IdentifyService {
 
     // if the peer record has been sent, prefer the addresses in the record as they are signed by the remote peer
     if (message.signedPeerRecord != null) {
-      log('received signedPeerRecord in push from %p', remotePeer)
+      log('received signedPeerRecord in push from %p', connection.remotePeer)
 
       let peerRecordEnvelope = message.signedPeerRecord
       const envelope = await RecordEnvelope.openAndCertify(peerRecordEnvelope, PeerRecord.DOMAIN)
@@ -448,7 +433,7 @@ export class DefaultIdentifyService implements Startable, IdentifyService {
       }
 
       // Make sure remote peer is the one sending the record
-      if (!remotePeer.equals(peerRecord.peerId)) {
+      if (!connection.remotePeer.equals(peerRecord.peerId)) {
         throw new Error('signing key does not match remote PeerId')
       }
 
@@ -494,7 +479,7 @@ export class DefaultIdentifyService implements Startable, IdentifyService {
         addresses: peerRecord.multiaddrs
       }
     } else {
-      log('%p did not send a signed peer record', remotePeer)
+      log('%p did not send a signed peer record', connection.remotePeer)
     }
 
     if (message.agentVersion != null) {
@@ -505,9 +490,23 @@ export class DefaultIdentifyService implements Startable, IdentifyService {
       peer.metadata.set('ProtocolVersion', uint8ArrayFromString(message.protocolVersion))
     }
 
-    await this.peerStore.patch(remotePeer, peer)
+    await this.peerStore.patch(connection.remotePeer, peer)
 
-    return output
+    const result: IdentifyResult = {
+      peerId: connection.remotePeer,
+      protocolVersion: message.protocolVersion,
+      agentVersion: message.agentVersion,
+      publicKey: message.publicKey,
+      listenAddrs: message.listenAddrs.map(buf => multiaddr(buf)),
+      observedAddr: message.observedAddr == null ? undefined : multiaddr(message.observedAddr),
+      protocols: message.protocols,
+      signedPeerRecord: output,
+      connection
+    }
+
+    this.events.safeDispatchEvent('peer:identify', { detail: result })
+
+    return result
   }
 }
 

packages/libp2p/src/registrar.ts

@@ -2,7 +2,7 @@ import { CodeError } from '@libp2p/interface/errors'
 import { logger } from '@libp2p/logger'
 import merge from 'merge-options'
 import { codes } from './errors.js'
-import type { Libp2pEvents, PeerUpdate } from '@libp2p/interface'
+import type { IdentifyResult, Libp2pEvents, PeerUpdate } from '@libp2p/interface'
 import type { TypedEventTarget } from '@libp2p/interface/events'
 import type { PeerId } from '@libp2p/interface/peer-id'
 import type { PeerStore } from '@libp2p/interface/peer-store'
@@ -37,11 +37,11 @@ export class DefaultRegistrar implements Registrar {
 
     this._onDisconnect = this._onDisconnect.bind(this)
     this._onPeerUpdate = this._onPeerUpdate.bind(this)
-    this._onConnect = this._onConnect.bind(this)
+    this._onPeerIdentify = this._onPeerIdentify.bind(this)
 
     this.components.events.addEventListener('peer:disconnect', this._onDisconnect)
-    this.components.events.addEventListener('peer:connect', this._onConnect)
     this.components.events.addEventListener('peer:update', this._onPeerUpdate)
+    this.components.events.addEventListener('peer:identify', this._onPeerIdentify)
   }
 
   getProtocols (): string[] {
@@ -183,52 +183,12 @@ export class DefaultRegistrar implements Registrar {
   }
 
   /**
-   * On peer connected if we already have their protocols. Usually used for reconnects
-   * as change:protocols event won't be emitted due to identical protocols.
-   */
-  _onConnect (evt: CustomEvent<PeerId>): void {
-    const remotePeer = evt.detail
-
-    void this.components.peerStore.get(remotePeer)
-      .then(peer => {
-        const connection = this.components.connectionManager.getConnections(peer.id)[0]
-
-        if (connection == null) {
-          log('peer %p connected but the connection manager did not have a connection', peer)
-          // peer disconnected while we were loading their details from the peer store
-          return
-        }
-
-        for (const protocol of peer.protocols) {
-          const topologies = this.topologies.get(protocol)
-
-          if (topologies == null) {
-            // no topologies are interested in this protocol
-            continue
-          }
-
-          for (const topology of topologies.values()) {
-            topology.onConnect?.(remotePeer, connection)
-          }
-        }
-      })
-      .catch(err => {
-        if (err.code === codes.ERR_NOT_FOUND) {
-          // peer has not completed identify so they are not in the peer store
-          return
-        }
-
-        log.error('could not inform topologies of connecting peer %p', remotePeer, err)
-      })
-  }
-
-  /**
-   * Check if a new peer support the multicodecs for this topology
+   * When a peer is updated, if they have removed supported protocols notify any
+   * topologies interested in the removed protocols.
    */
   _onPeerUpdate (evt: CustomEvent<PeerUpdate>): void {
     const { peer, previous } = evt.detail
     const removed = (previous?.protocols ?? []).filter(protocol => !peer.protocols.includes(protocol))
-    const added = peer.protocols.filter(protocol => !(previous?.protocols ?? []).includes(protocol))
 
     for (const protocol of removed) {
       const topologies = this.topologies.get(protocol)
@@ -242,8 +202,18 @@ export class DefaultRegistrar implements Registrar {
         topology.onDisconnect?.(peer.id)
       }
     }
+  }
 
-    for (const protocol of added) {
+  /**
+   * After identify has completed and we have received the list of supported
+   * protocols, notify any topologies interested in those protocols.
+   */
+  _onPeerIdentify (evt: CustomEvent<IdentifyResult>): void {
+    const protocols = evt.detail.protocols
+    const connection = evt.detail.connection
+    const peerId = evt.detail.peerId
+
+    for (const protocol of protocols) {
       const topologies = this.topologies.get(protocol)
 
       if (topologies == null) {
@@ -252,12 +222,11 @@ export class DefaultRegistrar implements Registrar {
       }
 
       for (const topology of topologies.values()) {
-        const connection = this.components.connectionManager.getConnections(peer.id)[0]
-
-        if (connection == null) {
+        if (connection.transient && topology.notifyOnTransient !== true) {
           continue
         }
-        topology.onConnect?.(peer.id, connection)
+
+        topology.onConnect?.(peerId, connection)
       }
     }
   }

packages/libp2p/test/identify/service.node.ts

@@ -65,13 +65,15 @@ describe('identify', () => {
     expect(connection).to.exist()
 
     // wait for identify to run on the new connection
-    await eventPromise
+    const identifyResult = await eventPromise
+
+    // should have run on the new connection
+    expect(identifyResult).to.have.nested.property('detail.connection', connection)
 
     // assert we have received certified announce addresses
-    const peer = await libp2p.peerStore.get(remoteLibp2p.peerId)
-    expect(peer.addresses).to.have.lengthOf(1)
-    expect(peer.addresses[0].isCertified).to.be.true('did not receive certified address via identify')
-    expect(peer.addresses[0].multiaddr.toString()).to.startWith('/dns4/localhost/', 'did not receive announce address via identify')
+    expect(identifyResult).to.have.deep.nested.property('detail.signedPeerRecord.addresses', [
+      multiaddr(`/dns4/localhost/tcp/${REMOTE_PORT}`)
+    ], 'did not receive announce address via identify')
   })
 
   it('should run identify automatically for inbound connections', async () => {
@@ -88,14 +90,16 @@ describe('identify', () => {
     const connection = await remoteLibp2p.dial(multiaddr(`/ip4/127.0.0.1/tcp/${LOCAL_PORT}/p2p/${libp2p.peerId.toString()}`))
     expect(connection).to.exist()
 
-    // wait for identify to run on the new connection
-    await eventPromise
+    // wait for identify to run
+    const identifyResult = await eventPromise
+
+    // should have run on the new connection
+    expect(identifyResult).to.have.nested.property('detail.connection', connection)
 
     // assert we have received certified announce addresses
-    const peer = await libp2p.peerStore.get(remoteLibp2p.peerId)
-    expect(peer.addresses).to.have.lengthOf(1)
-    expect(peer.addresses[0].isCertified).to.be.true('did not receive certified address via identify')
-    expect(peer.addresses[0].multiaddr.toString()).to.startWith('/dns4/localhost/', 'did not receive announce address via identify')
+    expect(identifyResult).to.have.deep.nested.property('detail.signedPeerRecord.addresses', [
+      multiaddr(`/dns4/localhost/tcp/${LOCAL_PORT}`)
+    ], 'did not receive announce address via identify')
   })
 
   it('should identify connection on dial and get proper announce addresses', async () => {