fix(libp2p): emit peer:discovered event on internal event bus (#2019)

To notifiy components that a new peer has been discovered, emit the
`peer:discovered` event on the event bus component.

Allows the auto-dialer to choose to dial newly discovered peers
immediately instead of waiting for the next autodial interval.

Author: achingbrain

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

@@ -2,7 +2,7 @@ import { logger } from '@libp2p/logger'
 import { PeerMap, PeerSet } from '@libp2p/peer-collections'
 import { toString as uint8ArrayToString } from 'uint8arrays/to-string'
 import { PeerJobQueue } from '../utils/peer-job-queue.js'
-import { AUTO_DIAL_CONCURRENCY, AUTO_DIAL_INTERVAL, AUTO_DIAL_MAX_QUEUE_LENGTH, AUTO_DIAL_PEER_RETRY_THRESHOLD, AUTO_DIAL_PRIORITY, LAST_DIAL_FAILURE_KEY, MIN_CONNECTIONS } from './constants.js'
+import { AUTO_DIAL_CONCURRENCY, AUTO_DIAL_DISCOVERED_PEERS_DEBOUNCE, AUTO_DIAL_INTERVAL, AUTO_DIAL_MAX_QUEUE_LENGTH, AUTO_DIAL_PEER_RETRY_THRESHOLD, AUTO_DIAL_PRIORITY, LAST_DIAL_FAILURE_KEY, MIN_CONNECTIONS } from './constants.js'
 import type { Libp2pEvents } from '@libp2p/interface'
 import type { EventEmitter } from '@libp2p/interface/events'
 import type { PeerStore } from '@libp2p/interface/peer-store'
@@ -18,6 +18,7 @@ interface AutoDialInit {
   autoDialPriority?: number
   autoDialInterval?: number
   autoDialPeerRetryThreshold?: number
+  autoDialDiscoveredPeersDebounce?: number
 }
 
 interface AutoDialComponents {
@@ -32,7 +33,8 @@ const defaultOptions = {
   autoDialConcurrency: AUTO_DIAL_CONCURRENCY,
   autoDialPriority: AUTO_DIAL_PRIORITY,
   autoDialInterval: AUTO_DIAL_INTERVAL,
-  autoDialPeerRetryThreshold: AUTO_DIAL_PEER_RETRY_THRESHOLD
+  autoDialPeerRetryThreshold: AUTO_DIAL_PEER_RETRY_THRESHOLD,
+  autoDialDiscoveredPeersDebounce: AUTO_DIAL_DISCOVERED_PEERS_DEBOUNCE
 }
 
 export class AutoDial implements Startable {
@@ -44,6 +46,7 @@ export class AutoDial implements Startable {
   private readonly autoDialIntervalMs: number
   private readonly autoDialMaxQueueLength: number
   private readonly autoDialPeerRetryThresholdMs: number
+  private readonly autoDialDiscoveredPeersDebounce: number
   private autoDialInterval?: ReturnType<typeof setInterval>
   private started: boolean
   private running: boolean
@@ -61,6 +64,7 @@ export class AutoDial implements Startable {
     this.autoDialIntervalMs = init.autoDialInterval ?? defaultOptions.autoDialInterval
     this.autoDialMaxQueueLength = init.maxQueueLength ?? defaultOptions.maxQueueLength
     this.autoDialPeerRetryThresholdMs = init.autoDialPeerRetryThreshold ?? defaultOptions.autoDialPeerRetryThreshold
+    this.autoDialDiscoveredPeersDebounce = init.autoDialDiscoveredPeersDebounce ?? defaultOptions.autoDialDiscoveredPeersDebounce
     this.started = false
     this.running = false
     this.queue = new PeerJobQueue({
@@ -77,6 +81,22 @@ export class AutoDial implements Startable {
           log.error(err)
         })
     })
+
+    // sometimes peers are discovered in quick succession so add a small
+    // debounce to ensure all eligible peers are autodialed
+    let debounce: ReturnType<typeof setTimeout>
+
+    // when new peers are discovered, dial them if we don't have
+    // enough connections
+    components.events.addEventListener('peer:discovery', () => {
+      clearTimeout(debounce)
+      debounce = setTimeout(() => {
+        this.autoDial()
+          .catch(err => {
+            log.error(err)
+          })
+      }, this.autoDialDiscoveredPeersDebounce)
+    })
   }
 
   isStarted (): boolean {

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

@@ -38,6 +38,11 @@ export const AUTO_DIAL_MAX_QUEUE_LENGTH = 100
  */
 export const AUTO_DIAL_PEER_RETRY_THRESHOLD = 1000 * 60
 
+/**
+ * @see https://libp2p.github.io/js-libp2p/interfaces/libp2p.index.unknown.ConnectionManagerInit.html#autoDialDiscoveredPeersDebounce
+ */
+export const AUTO_DIAL_DISCOVERED_PEERS_DEBOUNCE = 10
+
 /**
  * @see https://libp2p.github.io/js-libp2p/interfaces/index._internal_.ConnectionManagerConfig.html#inboundConnectionThreshold
  */

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

@@ -72,6 +72,14 @@ export interface ConnectionManagerInit {
    */
   autoDialPeerRetryThreshold?: number
 
+  /**
+   * Newly discovered peers may be auto-dialed to increase the number of open
+   * connections, but they can be discovered in quick succession so add a small
+   * delay before attempting to dial them in case more peers have been
+   * discovered. (default: 10ms)
+   */
+  autoDialDiscoveredPeersDebounce?: number
+
   /**
    * Sort the known addresses of a peer before trying to dial, By default public
    * addresses will be dialled before private (e.g. loopback or LAN) addresses.

packages/libp2p/src/libp2p.ts

@@ -105,7 +105,7 @@ export class Libp2pNode<T extends ServiceMap = Record<string, unknown>> extends
           protocols: evt.detail.peer.protocols
         }
 
-        this.safeDispatchEvent('peer:discovery', { detail: peerInfo })
+        components.events.safeDispatchEvent('peer:discovery', { detail: peerInfo })
       }
     })