feat: discover and connect to closest peers (#798)

Author: vasco-santos

doc/API.md

@@ -20,6 +20,7 @@
   * [`contentRouting.get`](#contentroutingget)
   * [`contentRouting.getMany`](#contentroutinggetmany)
   * [`peerRouting.findPeer`](#peerroutingfindpeer)
+  * [`peerRouting.getClosestPeers`](#peerroutinggetclosestpeers)
   * [`peerStore.addressBook.add`](#peerstoreaddressbookadd)
   * [`peerStore.addressBook.delete`](#peerstoreaddressbookdelete)
   * [`peerStore.addressBook.get`](#peerstoreaddressbookget)
@@ -99,6 +100,7 @@ Creates an instance of Libp2p.
 | [options.keychain] | [`object`](./CONFIGURATION.md#setup-with-keychain) | keychain [configuration](./CONFIGURATION.md#setup-with-keychain) |
 | [options.metrics] | [`object`](./CONFIGURATION.md#configuring-metrics) | libp2p Metrics [configuration](./CONFIGURATION.md#configuring-metrics) |
 | [options.peerId] | [`PeerId`][peer-id] | peerId instance (it will be created if not provided) |
+| [options.peerRouting] | [`object`](./CONFIGURATION.md#setup-with-content-and-peer-routing) | libp2p Peer routing service [configuration](./CONFIGURATION.md#setup-with-content-and-peer-routing) |
 | [options.peerStore] | [`object`](./CONFIGURATION.md#configuring-peerstore) | libp2p PeerStore [configuration](./CONFIGURATION.md#configuring-peerstore) |
 
 For Libp2p configurations and modules details read the [Configuration Document](./CONFIGURATION.md).
@@ -675,6 +677,36 @@ Iterates over all peer routers in series to find the given peer. If the DHT is e
 const peer = await libp2p.peerRouting.findPeer(peerId, options)
 ```
 
+### peerRouting.getClosestPeers
+
+Iterates over all content routers in series to get the closest peers of the given key.
+Once a content router succeeds, the iteration will stop. If the DHT is enabled, it will be queried first.
+
+`libp2p.peerRouting.getClosestPeers(cid, options)`
+
+#### Parameters
+
+| Name | Type | Description |
+|------|------|-------------|
+| key | `Uint8Array` | A CID like key |
+| options | `object` | operation options |
+| options.timeout | `number` | How long the query can take (ms). |
+
+#### Returns
+
+| Type | Description |
+|------|-------------|
+| `AsyncIterable<{ id: PeerId, multiaddrs: Multiaddr[] }` |  Async iterator for peer data |
+
+#### Example
+
+```js
+// Iterate over the closest peers found for the given key
+for await (const peer of libp2p.peerRouting.getClosestPeers(key)) {
+  console.log(peer.id, peer.multiaddrs)
+}
+```
+
 ### peerStore.addressBook.add
 
 Adds known `multiaddrs` of a given peer. If the peer is not known, it will be set with the provided multiaddrs.

doc/CONFIGURATION.md

@@ -397,7 +397,14 @@ const node = await Libp2p.create({
       new DelegatedPeerRouter()
     ],
   },
-  peerId
+  peerId,
+  peerRouting: { // Peer routing configuration
+    refreshManager: { // Refresh known and connected closest peers
+      enabled: true, // Should find the closest peers.
+      interval: 6e5, // Interval for getting the new for closest peers of 10min
+      bootDelay: 10e3 // Delay for the initial query for closest peers
+    }
+  }
 })
 ```
 

package.json

@@ -79,6 +79,7 @@
     "protons": "^2.0.0",
     "retimer": "^2.0.0",
     "sanitize-filename": "^1.6.3",
+    "set-delayed-interval": "^1.0.0",
     "streaming-iterables": "^5.0.2",
     "timeout-abort-controller": "^1.1.1",
     "varint": "^5.0.0",
@@ -92,6 +93,7 @@
     "chai-string": "^1.5.0",
     "delay": "^4.3.0",
     "interop-libp2p": "^0.3.0",
+    "into-stream": "^6.0.0",
     "ipfs-http-client": "^47.0.1",
     "it-concat": "^1.0.0",
     "it-pair": "^1.0.0",

src/config.js

@@ -37,6 +37,13 @@ const DefaultConfig = {
     persistence: false,
     threshold: 5
   },
+  peerRouting: {
+    refreshManager: {
+      enabled: true,
+      interval: 6e5,
+      bootDelay: 10e3
+    }
+  },
   config: {
     dht: {
       enabled: false,

src/index.js

@@ -9,7 +9,7 @@ log.error = debug('libp2p:error')
 const errCode = require('err-code')
 const PeerId = require('peer-id')
 
-const peerRouting = require('./peer-routing')
+const PeerRouting = require('./peer-routing')
 const contentRouting = require('./content-routing')
 const getPeer = require('./get-peer')
 const { validate: validateConfig } = require('./config')
@@ -193,7 +193,7 @@ class Libp2p extends EventEmitter {
 
     // Attach remaining APIs
     // peer and content routing will automatically get modules from _modules and _dht
-    this.peerRouting = peerRouting(this)
+    this.peerRouting = new PeerRouting(this)
     this.contentRouting = contentRouting(this)
 
     // Mount default protocols
@@ -250,8 +250,8 @@ class Libp2p extends EventEmitter {
     try {
       this._isStarted = false
 
-      // Relay
       this.relay && this.relay.stop()
+      this.peerRouting.stop()
 
       for (const service of this._discovery.values()) {
         service.removeListener('peer', this._onDiscoveryPeer)
@@ -496,6 +496,8 @@ class Libp2p extends EventEmitter {
 
     // Relay
     this.relay && this.relay.start()
+
+    this.peerRouting.start()
   }
 
   /**

src/peer-routing.js

@@ -1,40 +1,126 @@
 'use strict'
 
 const errCode = require('err-code')
+const debug = require('debug')
+const log = debug('libp2p:peer-routing')
+log.error = debug('libp2p:peer-routing:error')
+
+const all = require('it-all')
 const pAny = require('p-any')
+const {
+  setDelayedInterval,
+  clearDelayedInterval
+} = require('set-delayed-interval')
+
+/**
+ * Responsible for managing the usage of the available Peer Routing modules.
+ */
+class PeerRouting {
+  /**
+   * @class
+   * @param {Libp2p} libp2p
+   */
+  constructor (libp2p) {
+    this._peerId = libp2p.peerId
+    this._peerStore = libp2p.peerStore
+    this._routers = libp2p._modules.peerRouting || []
+
+    // If we have the dht, make it first
+    if (libp2p._dht) {
+      this._routers.unshift(libp2p._dht)
+    }
+
+    this._refreshManagerOptions = libp2p._options.peerRouting.refreshManager
+
+    this._findClosestPeersTask = this._findClosestPeersTask.bind(this)
+  }
+
+  /**
+   * Start peer routing service.
+   */
+  start () {
+    if (!this._routers.length || this._timeoutId || !this._refreshManagerOptions.enabled) {
+      return
+    }
+
+    this._timeoutId = setDelayedInterval(
+      this._findClosestPeersTask, this._refreshManagerOptions.interval, this._refreshManagerOptions.bootDelay
+    )
+  }
 
-module.exports = (node) => {
-  const routers = node._modules.peerRouting || []
+  /**
+   * Recurrent task to find closest peers and add their addresses to the Address Book.
+   */
+  async _findClosestPeersTask () {
+    try {
+      for await (const { id, multiaddrs } of this.getClosestPeers(this._peerId.id)) {
+        this._peerStore.addressBook.add(id, multiaddrs)
+      }
+    } catch (err) {
+      log.error(err)
+    }
+  }
 
-  // If we have the dht, make it first
-  if (node._dht) {
-    routers.unshift(node._dht)
+  /**
+   * Stop peer routing service.
+   */
+  stop () {
+    clearDelayedInterval(this._timeoutId)
   }
 
-  return {
-    /**
-     * Iterates over all peer routers in series to find the given peer.
-     *
-     * @param {string} id - The id of the peer to find
-     * @param {object} [options]
-     * @param {number} [options.timeout] - How long the query should run
-     * @returns {Promise<{ id: PeerId, multiaddrs: Multiaddr[] }>}
-     */
-    findPeer: async (id, options) => { // eslint-disable-line require-await
-      if (!routers.length) {
-        throw errCode(new Error('No peer routers available'), 'NO_ROUTERS_AVAILABLE')
+  /**
+   * Iterates over all peer routers in series to find the given peer.
+   *
+   * @param {string} id - The id of the peer to find
+   * @param {object} [options]
+   * @param {number} [options.timeout] - How long the query should run
+   * @returns {Promise<{ id: PeerId, multiaddrs: Multiaddr[] }>}
+   */
+  async findPeer (id, options) { // eslint-disable-line require-await
+    if (!this._routers.length) {
+      throw errCode(new Error('No peer routers available'), 'NO_ROUTERS_AVAILABLE')
+    }
+
+    return pAny(this._routers.map(async (router) => {
+      const result = await router.findPeer(id, options)
+
+      // If we don't have a result, we need to provide an error to keep trying
+      if (!result || Object.keys(result).length === 0) {
+        throw errCode(new Error('not found'), 'NOT_FOUND')
       }
 
-      return pAny(routers.map(async (router) => {
-        const result = await router.findPeer(id, options)
+      return result
+    }))
+  }
+
+  /**
+   * Attempt to find the closest peers on the network to the given key.
+   *
+   * @param {Uint8Array} key - A CID like key
+   * @param {Object} [options]
+   * @param {number} [options.timeout=30e3] - How long the query can take.
+   * @returns {AsyncIterable<{ id: PeerId, multiaddrs: Multiaddr[] }>}
+   */
+  async * getClosestPeers (key, options = { timeout: 30e3 }) {
+    if (!this._routers.length) {
+      throw errCode(new Error('No peer routers available'), 'NO_ROUTERS_AVAILABLE')
+    }
 
-        // If we don't have a result, we need to provide an error to keep trying
-        if (!result || Object.keys(result).length === 0) {
+    const result = await pAny(
+      this._routers.map(async (router) => {
+        const peers = await all(router.getClosestPeers(key, options))
+
+        if (!peers || !peers.length) {
           throw errCode(new Error('not found'), 'NOT_FOUND')
         }
+        return peers
+      })
+    )
 
-        return result
-      }))
+    for (const peer of result) {
+      yield peer
     }
   }
 }
+
+module.exports = PeerRouting