fix: dial peers until relays are discovered (#3188)

When no peer routing is configured the randomwalk performed to find
potential relay peers cannot be performed - to allow relay discovery
with these setups, dial every peer discovered before we reach the
specified number of relays, then identify will happen and the
network topology will kick in as normal.

Author: achingbrain

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

@@ -5,7 +5,7 @@ import { toString as uint8ArrayToString } from 'uint8arrays/to-string'
 import {
   RELAY_V2_HOP_CODEC
 } from '../constants.js'
-import type { ComponentLogger, Logger, Peer, PeerId, PeerStore, Startable, TopologyFilter } from '@libp2p/interface'
+import type { ComponentLogger, Libp2pEvents, Logger, Peer, PeerId, PeerInfo, PeerStore, Startable, TopologyFilter, TypedEventTarget } from '@libp2p/interface'
 import type { ConnectionManager, RandomWalk, Registrar, TransportManager } from '@libp2p/interface-internal'
 
 export interface RelayDiscoveryEvents {
@@ -19,6 +19,7 @@ export interface RelayDiscoveryComponents {
   registrar: Registrar
   logger: ComponentLogger
   randomWalk: RandomWalk
+  events: TypedEventTarget<Libp2pEvents>
 }
 
 export interface RelayDiscoveryInit {
@@ -30,10 +31,7 @@ export interface RelayDiscoveryInit {
  * peers that support the circuit v2 HOP protocol.
  */
 export class RelayDiscovery extends TypedEventEmitter<RelayDiscoveryEvents> implements Startable {
-  private readonly peerStore: PeerStore
-  private readonly registrar: Registrar
-  private readonly connectionManager: ConnectionManager
-  private readonly randomWalk: RandomWalk
+  private readonly components: RelayDiscoveryComponents
   private started: boolean
   private running: boolean
   private topologyId?: string
@@ -46,15 +44,14 @@ export class RelayDiscovery extends TypedEventEmitter<RelayDiscoveryEvents> impl
     super()
 
     this.log = components.logger.forComponent('libp2p:circuit-relay:discover-relays')
+    this.components = components
     this.started = false
     this.running = false
-    this.peerStore = components.peerStore
-    this.registrar = components.registrar
-    this.connectionManager = components.connectionManager
-    this.randomWalk = components.randomWalk
     this.filter = init.filter
     this.discoveryController = new AbortController()
     setMaxListeners(Infinity, this.discoveryController.signal)
+    this.dialPeer = this.dialPeer.bind(this)
+    this.onPeer = this.onPeer.bind(this)
   }
 
   isStarted (): boolean {
@@ -64,7 +61,7 @@ export class RelayDiscovery extends TypedEventEmitter<RelayDiscoveryEvents> impl
   async start (): Promise<void> {
     // 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, {
+    this.topologyId = await this.components.registrar.register(RELAY_V2_HOP_CODEC, {
       filter: this.filter,
       onConnect: (peerId) => {
         this.log.trace('discovered relay %p queue (length: %d, active %d)', peerId, this.queue?.size, this.queue?.running)
@@ -77,10 +74,13 @@ export class RelayDiscovery extends TypedEventEmitter<RelayDiscoveryEvents> impl
 
   stop (): void {
     if (this.topologyId != null) {
-      this.registrar.unregister(this.topologyId)
+      this.components.registrar.unregister(this.topologyId)
+    }
+
+    if (this.running) {
+      this.stopDiscovery()
     }
 
-    this.discoveryController?.abort()
     this.started = false
   }
 
@@ -90,7 +90,8 @@ export class RelayDiscovery extends TypedEventEmitter<RelayDiscoveryEvents> impl
    *
    * 1. Check the metadata store for known relays, try to listen on the ones we are already connected to
    * 2. Dial and try to listen on the peers we know that support hop but are not connected
-   * 3. Search the network
+   * 3. Search the network - this requires a peer routing implementation to be configured but will fail gracefully
+   * 4. Dial any peers discovered - this covers when no peer routing implementation has been configured but some peer discovery mechanism is also present
    */
   startDiscovery (): void {
     if (this.running) {
@@ -102,11 +103,14 @@ export class RelayDiscovery extends TypedEventEmitter<RelayDiscoveryEvents> impl
     this.discoveryController = new AbortController()
     setMaxListeners(Infinity, this.discoveryController.signal)
 
+    // dial any peer we discover
+    this.components.events.addEventListener('peer:discovery', this.onPeer)
+
     Promise.resolve()
       .then(async () => {
         this.log('searching peer store for relays')
 
-        const peers = (await this.peerStore.all({
+        const peers = (await this.components.peerStore.all({
           filters: [
             // filter by a list of peers supporting RELAY_V2_HOP and ones we are not listening on
             (peer) => {
@@ -149,7 +153,7 @@ export class RelayDiscovery extends TypedEventEmitter<RelayDiscoveryEvents> impl
 
         this.log('start random walk')
 
-        for await (const peer of this.randomWalk.walk({ signal: this.discoveryController.signal })) {
+        for await (const peer of this.components.randomWalk.walk({ signal: this.discoveryController.signal })) {
           this.log.trace('found random peer %p', peer.id)
 
           if (queue.has(peer.id)) {
@@ -159,14 +163,14 @@ export class RelayDiscovery extends TypedEventEmitter<RelayDiscoveryEvents> impl
             continue
           }
 
-          if (this.connectionManager.getConnections(peer.id)?.length > 0) {
+          if (this.components.connectionManager.getConnections(peer.id)?.length > 0) {
             this.log.trace('random peer %p was already connected', peer.id)
 
             // skip peers we are already connected to
             continue
           }
 
-          if (!(await this.connectionManager.isDialable(peer.multiaddrs))) {
+          if (!(await this.components.connectionManager.isDialable(peer.multiaddrs))) {
             this.log.trace('random peer %p was not dialable', peer.id, peer.multiaddrs.map(ma => ma.toString()))
 
             // skip peers we can't dial
@@ -186,16 +190,7 @@ export class RelayDiscovery extends TypedEventEmitter<RelayDiscoveryEvents> impl
 
           // dial the peer - this will cause identify to run and our topology to
           // be notified and we'll attempt to create reservations
-          queue.add(async () => {
-            const signal = anySignal([this.discoveryController.signal, AbortSignal.timeout(5000)])
-            setMaxListeners(Infinity, signal)
-
-            try {
-              await this.connectionManager.openConnection(peer.id, { signal })
-            } finally {
-              signal.clear()
-            }
-          }, {
+          queue.add(this.dialPeer, {
             peerId: peer.id,
             signal: this.discoveryController.signal
           })
@@ -219,6 +214,70 @@ export class RelayDiscovery extends TypedEventEmitter<RelayDiscoveryEvents> impl
     this.log('stop discovery')
     this.running = false
     this.discoveryController?.abort()
+    this.queue?.clear()
+
+    // stop dialing any peer we discover
+    this.components.events.removeEventListener('peer:discovery', this.onPeer)
+  }
+
+  onPeer (evt: CustomEvent<PeerInfo>): void {
+    this.log.trace('maybe dialing discovered peer %p - %e', evt.detail.id)
+
+    this.maybeDialPeer(evt)
+      .catch(err => {
+        this.log.trace('error dialing discovered peer %p - %e', evt.detail.id, err)
+      })
+  }
+
+  async maybeDialPeer (evt: CustomEvent<PeerInfo>): Promise<void> {
+    if (this.queue == null) {
+      return
+    }
+
+    const peerId = evt.detail.id
+    const multiaddrs = evt.detail.multiaddrs
+
+    if (this.queue.has(peerId)) {
+      this.log.trace('random peer %p was already in queue', peerId)
+
+      // skip peers already in the queue
+      return
+    }
+
+    if (this.components.connectionManager.getConnections(peerId)?.length > 0) {
+      this.log.trace('random peer %p was already connected', peerId)
+
+      // skip peers we are already connected to
+      return
+    }
+
+    if (!(await this.components.connectionManager.isDialable(multiaddrs))) {
+      this.log.trace('random peer %p was not dialable', peerId)
+
+      // skip peers we can't dial
+      return
+    }
+
+    this.queue?.add(this.dialPeer, {
+      peerId: evt.detail.id,
+      signal: this.discoveryController.signal
+    })
+      .catch(err => {
+        this.log.error('error opening connection to discovered peer %p', evt.detail.id, err)
+      })
+  }
+
+  async dialPeer ({ peerId, signal }: { peerId: PeerId, signal?: AbortSignal }): Promise<void> {
+    const combinedSignal = anySignal([AbortSignal.timeout(5_000), signal])
+    setMaxListeners(Infinity, combinedSignal)
+
+    try {
+      await this.components.connectionManager.openConnection(peerId, {
+        signal: combinedSignal
+      })
+    } finally {
+      combinedSignal.clear()
+    }
   }
 }