fix: do not pollute routing table with useless peers (#254)

If a remote peer doesn't have private addresses, do not add it to the
routing table for the lan DHT, if they don't have public addresses,
do not add them to the routing table for the wan DHT.

Also adds more logging.

Author: achingbrain

src/kad-dht.js

@@ -257,7 +257,8 @@ class KadDHT extends EventEmitter {
     })
     this._topologyListener = new TopologyListener({
       registrar: libp2p.registrar,
-      protocol: this._protocol
+      protocol: this._protocol,
+      lan
     })
     this._querySelf = new QuerySelf({
       peerId: libp2p.peerId,
@@ -268,17 +269,22 @@ class KadDHT extends EventEmitter {
 
     // handle peers being discovered during processing of DHT messages
     this._network.on('peer', (peerData) => {
-      this._routingTable.add(peerData.id).catch(err => {
-        this._log.error(`Could not add ${peerData.id} to routing table`, err)
+      this.onPeerConnect(peerData).catch(err => {
+        this._log.error(`could not add ${peerData.id} to routing table`, err)
       })
 
       this.emit('peer', peerData)
     })
 
     // handle peers being discovered via other peer discovery mechanisms
     this._topologyListener.on('peer', async (peerId) => {
-      this._routingTable.add(peerId).catch(err => {
-        this._log.error(`Could not add ${peerId} to routing table`, err)
+      const peerData = {
+        id: peerId,
+        multiaddrs: (this._libp2p.peerStore.addressBook.get(peerId) || []).map((/** @type {{ multiaddr: Multiaddr }} */ addr) => addr.multiaddr)
+      }
+
+      this.onPeerConnect(peerData).catch(err => {
+        this._log.error(`could not add ${peerData.id} to routing table`, err)
       })
     })
   }
@@ -287,24 +293,23 @@ class KadDHT extends EventEmitter {
    * @param {PeerData} peerData
    */
   async onPeerConnect (peerData) {
+    this._log('peer %p connected', peerData.id)
+
     if (this._lan) {
       peerData = removePublicAddresses(peerData)
     } else {
       peerData = removePrivateAddresses(peerData)
     }
 
     if (!peerData.multiaddrs.length) {
+      this._log('ignoring %p as they do not have any %s addresses in %s', peerData.id, this._lan ? 'private' : 'public', peerData.multiaddrs.map(addr => addr.toString()))
       return
     }
 
     try {
-      const has = await this._routingTable.find(peerData.id)
-
-      if (!has) {
-        await this._routingTable.add(peerData.id)
-      }
+      await this._routingTable.add(peerData.id)
     } catch (/** @type {any} */ err) {
-      this._log.error('Could not add %p to routing table', peerData.id, err)
+      this._log.error('could not add %p to routing table', peerData.id, err)
     }
   }
 

src/providers.js

@@ -180,7 +180,7 @@ class Providers {
    */
   async addProvider (cid, provider) { // eslint-disable-line require-await
     return this.syncQueue.add(async () => {
-      log('addProvider %s', cid.toString())
+      log('%p provides %s', provider, cid)
       const provs = await this._getProvidersMap(cid)
 
       log('loaded %s provs', provs.size)
@@ -202,7 +202,7 @@ class Providers {
    */
   async getProviders (cid) { // eslint-disable-line require-await
     return this.syncQueue.add(async () => {
-      log('getProviders %s', cid.toString())
+      log('get providers for %s', cid)
       const provs = await this._getProvidersMap(cid)
 
       return [...provs.keys()].map(peerIdStr => {

src/routing-table/index.js

@@ -83,15 +83,15 @@ class RoutingTable {
 
             try {
               timeoutController = new TimeoutController(this._pingTimeout)
-              this._log(`Pinging old contact ${oldContact.peer}`)
+              this._log(`pinging old contact ${oldContact.peer}`)
               const { stream } = await this._dialer.dialProtocol(oldContact.peer, PROTOCOL_DHT, {
                 signal: timeoutController.signal
               })
               await stream.close()
               responded++
             } catch (/** @type {any} */ err) {
-              this._log.error('Could not ping peer %p', oldContact.peer, err)
-              this._log(`Evicting old contact after ping failed ${oldContact.peer}`)
+              this._log.error('could not ping peer %p', oldContact.peer, err)
+              this._log(`evicting old contact after ping failed ${oldContact.peer}`)
               this.kb.remove(oldContact.id)
             } finally {
               if (timeoutController) {
@@ -102,11 +102,11 @@ class RoutingTable {
         )
 
         if (responded < oldContacts.length) {
-          this._log(`Adding new contact ${newContact.peer}`)
+          this._log(`adding new contact ${newContact.peer}`)
           this.kb.add(newContact)
         }
       } catch (/** @type {any} */ err) {
-        this._log.error('Could not process k-bucket ping event', err)
+        this._log.error('could not process k-bucket ping event', err)
       }
     })
   }
@@ -168,6 +168,8 @@ class RoutingTable {
     const id = await utils.convertPeerId(peer)
 
     this.kb.add({ id: id, peer: peer })
+
+    this._log('added %p with kad id %b', peer, id)
   }
 
   /**

src/rpc/handlers/get-providers.js

@@ -51,7 +51,7 @@ class GetProvidersHandler {
       throw errcode(new Error(`Invalid CID: ${err.message}`), 'ERR_INVALID_CID')
     }
 
-    log('%p asking for providers for %s', peerId, cid.toString())
+    log('%p asking for providers for %s', peerId, cid)
 
     const [peers, closer] = await Promise.all([
       this._providers.getProviders(cid),

src/topology-listener.js

@@ -2,6 +2,7 @@
 
 const MulticodecTopology = require('libp2p-interfaces/src/topology/multicodec-topology')
 const { EventEmitter } = require('events')
+const utils = require('./utils')
 
 /**
  * Receives notifications of new peers joining the network that support the DHT protocol
@@ -13,10 +14,12 @@ class TopologyListener extends EventEmitter {
    * @param {object} params
    * @param {import('./types').Registrar} params.registrar
    * @param {string} params.protocol
+   * @param {boolean} params.lan
    */
-  constructor ({ registrar, protocol }) {
+  constructor ({ registrar, protocol, lan }) {
     super()
 
+    this._log = utils.logger(`libp2p:kad-dht:topology-listener:${lan ? 'lan' : 'wan'}:network`)
     this._running = false
     this._registrar = registrar
     this._protocol = protocol
@@ -37,6 +40,7 @@ class TopologyListener extends EventEmitter {
       multicodecs: [this._protocol],
       handlers: {
         onConnect: (peerId) => {
+          this._log('observed peer that with protocol %s %p', this._protocol, peerId)
           this.emit('peer', peerId)
         },
         onDisconnect: () => {}

test/utils/test-dht.js

@@ -140,6 +140,10 @@ class TestDHT {
     const [c0, c1] = ConnectionPair()
     const routingTableChecks = []
 
+    // Libp2p dial adds multiaddrs to the addressBook
+    dhtA._libp2p.peerStore.addressBook.add(dhtB._libp2p.peerId, dhtB._libp2p.multiaddrs)
+    dhtB._libp2p.peerStore.addressBook.add(dhtA._libp2p.peerId, dhtA._libp2p.multiaddrs)
+
     // Notice peers of connection
     if (!dhtB._clientMode) {
       // B is a server, trigger connect events on A
@@ -171,10 +175,6 @@ class TestDHT {
       })
     }
 
-    // Libp2p dial adds multiaddrs to the addressBook
-    dhtA._libp2p.peerStore.addressBook.add(dhtB._libp2p.peerId, dhtB._libp2p.multiaddrs)
-    dhtB._libp2p.peerStore.addressBook.add(dhtA._libp2p.peerId, dhtA._libp2p.multiaddrs)
-
     // Check routing tables
     return Promise.all(routingTableChecks.map(check => {
       return pRetry(check, { retries: 50 })