chore: remove kad-dht, bootstrap and mdns deps from libp2p (#2313)

Refactors the tests for these modules to test use of the interfaces.

This will let us, for example, make a breaking change to `@libp2p/kad-dht` without it causing a major for libp2p.

A followup that moves the specific tests to the integration suite will come next.

Author: achingbrain

packages/interface/package.json

@@ -67,6 +67,7 @@
     "it-pushable": "^3.2.1",
     "it-stream-types": "^2.0.1",
     "multiformats": "^12.1.3",
+    "progress-events": "^1.0.0",
     "uint8arraylist": "^2.4.3"
   },
   "devDependencies": {

packages/interface/src/content-routing/index.ts

@@ -1,4 +1,4 @@
-import type { AbortOptions } from '../index.js'
+import type { RoutingOptions } from '../index.js'
 import type { PeerInfo } from '../peer-info/index.js'
 import type { CID } from 'multiformats/cid'
 
@@ -23,6 +23,14 @@ import type { CID } from 'multiformats/cid'
  */
 export const contentRoutingSymbol = Symbol.for('@libp2p/content-routing')
 
+/**
+ * Implementers of this interface can provide a ContentRouting implementation to
+ * interested callers.
+ */
+export interface ContentRoutingProvider {
+  [contentRoutingSymbol]: ContentRouting
+}
+
 export interface ContentRouting {
   /**
    * The implementation of this method should ensure that network peers know the
@@ -35,7 +43,7 @@ export interface ContentRouting {
    * await contentRouting.provide(cid)
    * ```
    */
-  provide(cid: CID, options?: AbortOptions): Promise<void>
+  provide(cid: CID, options?: RoutingOptions): Promise<void>
 
   /**
    * Find the providers of the passed CID.
@@ -49,7 +57,7 @@ export interface ContentRouting {
    * }
    * ```
    */
-  findProviders(cid: CID, options?: AbortOptions): AsyncIterable<PeerInfo>
+  findProviders(cid: CID, options?: RoutingOptions): AsyncIterable<PeerInfo>
 
   /**
    * Puts a value corresponding to the passed key in a way that can later be
@@ -65,7 +73,7 @@ export interface ContentRouting {
    * await contentRouting.put(key, value)
    * ```
    */
-  put(key: Uint8Array, value: Uint8Array, options?: AbortOptions): Promise<void>
+  put(key: Uint8Array, value: Uint8Array, options?: RoutingOptions): Promise<void>
 
   /**
    * Retrieves a value from the network corresponding to the passed key.
@@ -79,5 +87,5 @@ export interface ContentRouting {
    * const value = await contentRouting.get(key)
    * ```
    */
-  get(key: Uint8Array, options?: AbortOptions): Promise<Uint8Array>
+  get(key: Uint8Array, options?: RoutingOptions): Promise<Uint8Array>
 }

packages/interface/src/index.ts

@@ -27,6 +27,7 @@ import type { StreamHandler, StreamHandlerOptions } from './stream-handler/index
 import type { Topology } from './topology/index.js'
 import type { Listener } from './transport/index.js'
 import type { Multiaddr } from '@multiformats/multiaddr'
+import type { ProgressOptions } from 'progress-events'
 
 /**
  * Used by the connection manager to sort addresses into order before dialling
@@ -671,6 +672,28 @@ export type RecursivePartial<T> = {
   [P in keyof T]?: T[P] extends Array<infer I> ? Array<RecursivePartial<I>> : T[P] extends (...args: any[]) => any ? T[P] : RecursivePartial<T[P]>
 }
 
+/**
+ * When a routing operation involves reading values, these options allow
+ * controlling where the values are read from. By default libp2p will check
+ * local caches but may not use the network if a valid local value is found,
+ * these options allow tuning that behaviour.
+ */
+export interface RoutingOptions extends AbortOptions, ProgressOptions {
+  /**
+   * Pass `false` to not use the network
+   *
+   * @default true
+   */
+  useNetwork?: boolean
+
+  /**
+   * Pass `false` to not use cached values
+   *
+   * @default true
+   */
+  useCache?: boolean
+}
+
 export * from './connection/index.js'
 export * from './connection-encrypter/index.js'
 export * from './connection-gater/index.js'

packages/interface/src/peer-discovery/index.ts

@@ -22,6 +22,14 @@ import type { PeerInfo } from '../peer-info/index.js'
  */
 export const peerDiscoverySymbol = Symbol.for('@libp2p/peer-discovery')
 
+/**
+ * Implementers of this interface can provide a PeerDiscovery implementation to
+ * interested callers.
+ */
+export interface PeerDiscoveryProvider {
+  [peerDiscoverySymbol]: PeerDiscovery
+}
+
 export interface PeerDiscoveryEvents {
   'peer': CustomEvent<PeerInfo>
 }

packages/interface/src/peer-routing/index.ts

@@ -1,4 +1,4 @@
-import type { AbortOptions } from '../index.js'
+import type { RoutingOptions } from '../index.js'
 import type { PeerId } from '../peer-id/index.js'
 import type { PeerInfo } from '../peer-info/index.js'
 
@@ -23,6 +23,14 @@ import type { PeerInfo } from '../peer-info/index.js'
  */
 export const peerRoutingSymbol = Symbol.for('@libp2p/peer-routing')
 
+/**
+ * Implementers of this interface can provide a PeerRouting implementation to
+ * interested callers.
+ */
+export interface PeerRoutingProvider {
+  [peerRoutingSymbol]: PeerRouting
+}
+
 export interface PeerRouting {
   /**
    * Searches the network for peer info corresponding to the passed peer id.
@@ -34,7 +42,7 @@ export interface PeerRouting {
    * const peer = await peerRouting.findPeer(peerId, options)
    * ```
    */
-  findPeer(peerId: PeerId, options?: AbortOptions): Promise<PeerInfo>
+  findPeer(peerId: PeerId, options?: RoutingOptions): Promise<PeerInfo>
 
   /**
    * Search the network for peers that are closer to the passed key. Peer
@@ -49,5 +57,5 @@ export interface PeerRouting {
    * }
    * ```
    */
-  getClosestPeers(key: Uint8Array, options?: AbortOptions): AsyncIterable<PeerInfo>
+  getClosestPeers(key: Uint8Array, options?: RoutingOptions): AsyncIterable<PeerInfo>
 }

packages/libp2p/.aegir.js

@@ -8,7 +8,6 @@ export default {
       // use dynamic import because we only want to reference these files during the test run, e.g. after building
       const { webSockets } = await import('@libp2p/websockets')
       const { mplex } = await import('@libp2p/mplex')
-      const { noise } = await import('@chainsafe/libp2p-noise')
       const { createEd25519PeerId } = await import('@libp2p/peer-id-factory')
       const { yamux } = await import('@chainsafe/libp2p-yamux')
       const { WebSockets } = await import('@multiformats/mafmt')
@@ -39,7 +38,6 @@ export default {
           mplex()
         ],
         connectionEncryption: [
-          noise(),
           plaintext()
         ],
         services: {

packages/libp2p/package.json

@@ -103,10 +103,9 @@
     "interface-datastore": "^8.2.0",
     "it-all": "^3.0.2",
     "it-drain": "^3.0.2",
-    "it-filter": "^3.0.1",
-    "it-first": "^3.0.3",
     "it-map": "^3.0.4",
     "it-merge": "^3.0.0",
+    "it-parallel": "^3.0.6",
     "it-pipe": "^3.0.1",
     "it-stream-types": "^2.0.1",
     "merge-options": "^3.0.4",
@@ -119,14 +118,10 @@
     "uint8arrays": "^5.0.0"
   },
   "devDependencies": {
-    "@chainsafe/libp2p-noise": "^14.0.0",
     "@chainsafe/libp2p-yamux": "^6.0.1",
-    "@libp2p/bootstrap": "^10.0.7",
     "@libp2p/circuit-relay-v2": "^1.0.7",
     "@libp2p/identify": "^1.0.6",
     "@libp2p/interface-compliance-tests": "^5.0.7",
-    "@libp2p/kad-dht": "^11.0.7",
-    "@libp2p/mdns": "^10.0.7",
     "@libp2p/mplex": "^10.0.7",
     "@libp2p/plaintext": "^1.0.7",
     "@libp2p/tcp": "^9.0.7",

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

@@ -294,17 +294,14 @@ export class DialQueue {
 
   private createDialAbortControllers (userSignal?: AbortSignal): ClearableSignal {
     // let any signal abort the dial
-    const signal = anySignal(
-      [AbortSignal.timeout(this.dialTimeout),
-        this.shutDownController.signal,
-        userSignal
-      ]
-    )
-
-    try {
-      // This emitter gets listened to a lot
-      setMaxListeners?.(Infinity, signal)
-    } catch {}
+    const signal = anySignal([
+      AbortSignal.timeout(this.dialTimeout),
+      this.shutDownController.signal,
+      userSignal
+    ])
+
+    // This emitter gets listened to a lot
+    setMaxListeners(Infinity, signal)
 
     return signal
   }

packages/libp2p/src/content-routing/index.ts renamed to packages/libp2p/src/content-routing.ts

@@ -1,13 +1,9 @@
 import { CodeError } from '@libp2p/interface'
+import { PeerSet } from '@libp2p/peer-collections'
 import merge from 'it-merge'
-import { pipe } from 'it-pipe'
-import { codes, messages } from '../errors.js'
-import {
-  storeAddresses,
-  uniquePeers,
-  requirePeers
-} from './utils.js'
-import type { AbortOptions, ContentRouting, PeerInfo, PeerStore, Startable } from '@libp2p/interface'
+import parallel from 'it-parallel'
+import { codes, messages } from './errors.js'
+import type { AbortOptions, ComponentLogger, ContentRouting, Logger, PeerInfo, PeerRouting, PeerStore, RoutingOptions, Startable } from '@libp2p/interface'
 import type { CID } from 'multiformats/cid'
 
 export interface CompoundContentRoutingInit {
@@ -16,14 +12,18 @@ export interface CompoundContentRoutingInit {
 
 export interface CompoundContentRoutingComponents {
   peerStore: PeerStore
+  peerRouting: PeerRouting
+  logger: ComponentLogger
 }
 
 export class CompoundContentRouting implements ContentRouting, Startable {
+  private readonly log: Logger
   private readonly routers: ContentRouting[]
   private started: boolean
   private readonly components: CompoundContentRoutingComponents
 
   constructor (components: CompoundContentRoutingComponents, init: CompoundContentRoutingInit) {
+    this.log = components.logger.forComponent('libp2p:content-routing')
     this.routers = init.routers ?? []
     this.started = false
     this.components = components
@@ -44,19 +44,65 @@ export class CompoundContentRouting implements ContentRouting, Startable {
   /**
    * Iterates over all content routers in parallel to find providers of the given key
    */
-  async * findProviders (key: CID, options: AbortOptions = {}): AsyncIterable<PeerInfo> {
+  async * findProviders (key: CID, options: RoutingOptions = {}): AsyncIterable<PeerInfo> {
     if (this.routers.length === 0) {
       throw new CodeError('No content routers available', codes.ERR_NO_ROUTERS_AVAILABLE)
     }
 
-    yield * pipe(
-      merge(
-        ...this.routers.map(router => router.findProviders(key, options))
-      ),
-      (source) => storeAddresses(source, this.components.peerStore),
-      (source) => uniquePeers(source),
-      (source) => requirePeers(source)
-    )
+    const self = this
+    const seen = new PeerSet()
+
+    for await (const peer of parallel(
+      async function * () {
+        const source = merge(
+          ...self.routers.map(router => router.findProviders(key, options))
+        )
+
+        for await (let peer of source) {
+          yield async () => {
+            // find multiaddrs if they are missing
+            if (peer.multiaddrs.length === 0) {
+              try {
+                peer = await self.components.peerRouting.findPeer(peer.id, {
+                  ...options,
+                  useCache: false
+                })
+              } catch (err) {
+                self.log.error('could not find peer multiaddrs', err)
+                return
+              }
+            }
+
+            return peer
+          }
+        }
+      }()
+    )) {
+      // the peer was yielded by a content router without multiaddrs and we
+      // failed to load them
+      if (peer == null) {
+        continue
+      }
+
+      // skip peers without addresses
+      if (peer.multiaddrs.length === 0) {
+        continue
+      }
+
+      // ensure we have the addresses for a given peer
+      await this.components.peerStore.merge(peer.id, {
+        multiaddrs: peer.multiaddrs
+      })
+
+      // deduplicate peers
+      if (seen.has(peer.id)) {
+        continue
+      }
+
+      seen.add(peer.id)
+
+      yield peer
+    }
   }
 
   /**
@@ -68,7 +114,9 @@ export class CompoundContentRouting implements ContentRouting, Startable {
       throw new CodeError('No content routers available', codes.ERR_NO_ROUTERS_AVAILABLE)
     }
 
-    await Promise.all(this.routers.map(async (router) => { await router.provide(key, options) }))
+    await Promise.all(this.routers.map(async (router) => {
+      await router.provide(key, options)
+    }))
   }
 
   /**