feat: ping old DHT peers before eviction (#229)

Updates the DHT contact logic to ensure we can still connect to old peers when buckets are full and only evict if the peer does not respond.

Author: achingbrain

package.json

@@ -69,6 +69,7 @@
     "peer-id": "^0.15.0",
     "protobufjs": "^6.10.2",
     "streaming-iterables": "^6.0.0",
+    "timeout-abort-controller": "^1.1.1",
     "uint8arrays": "^3.0.0",
     "varint": "^6.0.0"
   },

src/routing-table/index.js

@@ -14,6 +14,10 @@ const log = Object.assign(debug('libp2p:dht:routing-table'), {
 })
 // @ts-ignore
 const length = require('it-length')
+const { default: Queue } = require('p-queue')
+const { PROTOCOL_DHT } = require('../constants')
+// @ts-expect-error no types
+const TimeoutController = require('timeout-abort-controller')
 
 /**
  * @typedef {object} KBucketPeer
@@ -54,12 +58,14 @@ class RoutingTable {
    * @param {object} [options]
    * @param {number} [options.kBucketSize=20]
    * @param {number} [options.refreshInterval=30000]
+   * @param {number} [options.pingTimeout=10000]
    */
-  constructor (dht, { kBucketSize, refreshInterval } = {}) {
+  constructor (dht, { kBucketSize, refreshInterval, pingTimeout } = {}) {
     this.peerId = dht.peerId
     this.dht = dht
     this._kBucketSize = kBucketSize || 20
     this._refreshInterval = refreshInterval || 30000
+    this._pingTimeout = pingTimeout || 10000
 
     /** @type {KBucketTree} */
     this.kb = new KBuck({
@@ -72,6 +78,7 @@ class RoutingTable {
 
     this._refreshTable = this._refreshTable.bind(this)
     this._onPing = this._onPing.bind(this)
+    this._pingQueue = new Queue({ concurrency: 1 })
   }
 
   async start () {
@@ -85,6 +92,8 @@ class RoutingTable {
     if (this._refreshTimeoutId) {
       clearTimeout(this._refreshTimeoutId)
     }
+
+    this._pingQueue.clear()
   }
 
   /**
@@ -295,26 +304,58 @@ class RoutingTable {
   }
 
   /**
-   * Called on the `ping` event from `k-bucket`.
-   * Currently this just removes the oldest contact from
-   * the list, without actually pinging the individual peers.
-   * This is the same as go does, but should probably
-   * be upgraded to actually ping the individual peers.
+   * Called on the `ping` event from `k-bucket` when a bucket is full
+   * and cannot split.
+   *
+   * `oldContacts.length` is defined by the `numberOfNodesToPing` param
+   * passed to the `k-bucket` constructor.
+   *
+   * `oldContacts` will not be empty and is the list of contacts that
+   * have not been contacted for the longest.
    *
    * @param {KBucketPeer[]} oldContacts
    * @param {KBucketPeer} newContact
    */
   _onPing (oldContacts, newContact) {
-    // just use the first one (k-bucket sorts from oldest to newest)
-    const oldest = oldContacts[0]
-
-    if (oldest) {
-      // remove the oldest one
-      this.kb.remove(oldest.id)
-    }
+    // add to a queue so multiple ping requests do not overlap and we don't
+    // flood the network with ping requests if lots of newContact requests
+    // are received
+    this._pingQueue.add(async () => {
+      let responded = 0
+
+      try {
+        await Promise.all(
+          oldContacts.map(async oldContact => {
+            let timeoutController
+
+            try {
+              timeoutController = new TimeoutController(this._pingTimeout)
+              log(`Pinging old contact ${oldContact.peer.toB58String()}`)
+              const conn = await this.dht.libp2p.dialProtocol(oldContact.peer, PROTOCOL_DHT, {
+                signal: timeoutController.signal
+              })
+              await conn.close()
+              responded++
+            } catch (err) {
+              log.error('Could not ping peer', err)
+              log(`Evicting old contact after ping failed ${oldContact.peer.toB58String()}`)
+              this.kb.remove(oldContact.id)
+            } finally {
+              if (timeoutController) {
+                timeoutController.clear()
+              }
+            }
+          })
+        )
 
-    // add the new one
-    this.kb.add(newContact)
+        if (responded < oldContacts.length) {
+          log(`Adding new contact ${newContact.peer.toB58String()}`)
+          this.kb.add(newContact)
+        }
+      } catch (err) {
+        log.error('Could not process k-bucket ping event', err)
+      }
+    })
   }
 
   // -- Public Interface

test/routing-table.spec.js

@@ -4,19 +4,30 @@
 const { expect } = require('aegir/utils/chai')
 const PeerId = require('peer-id')
 const random = require('lodash.random')
+const sinon = require('sinon')
 
 const RoutingTable = require('../src/routing-table')
 const kadUtils = require('../src/utils')
 const createPeerId = require('./utils/create-peer-id')
+const { PROTOCOL_DHT } = require('../src/constants')
 
 describe('Routing Table', () => {
   let table
 
   beforeEach(async function () {
     this.timeout(20 * 1000)
 
-    const id = await PeerId.create({ bits: 512 })
-    table = new RoutingTable(id, 20)
+    const dht = {
+      peerId: await PeerId.create({ bits: 512 }),
+      libp2p: {
+        dialProtocol: sinon.stub()
+      }
+    }
+
+    table = new RoutingTable(dht, {
+      kBucketSize: 20,
+      refreshInterval: 30000
+    })
   })
 
   it('add', async function () {
@@ -73,4 +84,94 @@ describe('Routing Table', () => {
     const key = await kadUtils.convertPeerId(peers[2])
     expect(table.closestPeers(key, 15)).to.have.length(15)
   })
+
+  it('favours old peers that respond to pings', async () => {
+    let fn
+
+    // execute queued functions immediately
+    table._pingQueue = {
+      add: (f) => {
+        fn = f
+      }
+    }
+
+    const peerIds = [
+      PeerId.createFromB58String('QmYobx1VAHP7Mi88LcDvLeQoWcc1Aa2rynYHpdEPBqHZi5'),
+      PeerId.createFromB58String('QmYobx1VAHP7Mi88LcDvLeQoWcc1Aa2rynYHpdEPBqHZi6')
+    ]
+
+    const oldPeer = {
+      id: peerIds[0].toBytes(),
+      peer: peerIds[0]
+    }
+    const newPeer = {
+      id: peerIds[1].toBytes(),
+      peer: peerIds[1]
+    }
+
+    table._onPing([oldPeer], newPeer)
+
+    // add the old peer
+    table.kb.add(oldPeer)
+
+    // simulate connection succeeding
+    table.dht.libp2p.dialProtocol.withArgs(oldPeer.peer, PROTOCOL_DHT).resolves({ close: sinon.stub() })
+
+    // perform the ping
+    await fn()
+
+    expect(table.dht.libp2p.dialProtocol.callCount).to.equal(1)
+    expect(table.dht.libp2p.dialProtocol.calledWith(oldPeer.peer)).to.be.true()
+
+    // did not add the new peer
+    expect(table.kb.get(newPeer.id)).to.be.null()
+
+    // kept the old peer
+    expect(table.kb.get(oldPeer.id)).to.not.be.null()
+  })
+
+  it('evicts oldest peer that does not respond to ping', async () => {
+    let fn
+
+    // execute queued functions immediately
+    table._pingQueue = {
+      add: (f) => {
+        fn = f
+      }
+    }
+
+    const peerIds = [
+      PeerId.createFromB58String('QmYobx1VAHP7Mi88LcDvLeQoWcc1Aa2rynYHpdEPBqHZi5'),
+      PeerId.createFromB58String('QmYobx1VAHP7Mi88LcDvLeQoWcc1Aa2rynYHpdEPBqHZi6')
+    ]
+
+    const oldPeer = {
+      id: peerIds[0].toBytes(),
+      peer: peerIds[0]
+    }
+    const newPeer = {
+      id: peerIds[1].toBytes(),
+      peer: peerIds[1]
+    }
+
+    table._onPing([oldPeer], newPeer)
+
+    // add the old peer
+    table.kb.add(oldPeer)
+
+    // libp2p fails to dial the old peer
+    table.dht.libp2p.dialProtocol = sinon.stub().withArgs(oldPeer.peer, PROTOCOL_DHT).rejects(new Error('Could not dial peer'))
+
+    // perform the ping
+    await fn()
+
+    expect(table.dht.libp2p.dialProtocol.callCount).to.equal(1)
+    expect(table.dht.libp2p.dialProtocol.calledWith(oldPeer.peer)).to.be.true()
+
+    // added the new peer
+    expect(table.kb.get(newPeer.id)).to.not.be.null()
+
+    // evicted the old peer
+    expect(table.kb.get(oldPeer.id)).to.be.null()
+  })
 })