chore: guard on table refresh running (#257)

We queue up ping requests to see if we should evict peers from the
routing table - if we are being shut down the requests will be
aborted so check to make sure we are still running before starting
another.

Also, don't evict a peer because the request failed if we aren't
running as it may have been cancelled which doesn't mean the remote
peer is invalid.

Author: achingbrain

src/routing-table/index.js

@@ -35,24 +35,29 @@ class RoutingTable {
     this._pingTimeout = pingTimeout || 10000
 
     /** @type {KBucketTree} */
-    this.kb = new KBuck({
-      numberOfNodesPerKBucket: this._kBucketSize,
-      numberOfNodesToPing: 1
-    })
+    this.kb // eslint-disable-line no-unused-expressions
 
     /** @type {Date[]} */
     this.commonPrefixLengthRefreshedAt = []
 
     this._onPing = this._onPing.bind(this)
     this._pingQueue = new Queue({ concurrency: 1 })
+    this._running = false
   }
 
   async start () {
-    this.kb.localNodeId = await utils.convertPeerId(this._peerId)
+    this._running = true
+
+    this.kb = new KBuck({
+      localNodeId: await utils.convertPeerId(this._peerId),
+      numberOfNodesPerKBucket: this._kBucketSize,
+      numberOfNodesToPing: 1
+    })
     this.kb.on('ping', this._onPing)
   }
 
   async stop () {
+    this._running = false
     this._pingQueue.clear()
   }
 
@@ -74,6 +79,10 @@ class RoutingTable {
     // flood the network with ping requests if lots of newContact requests
     // are received
     this._pingQueue.add(async () => {
+      if (!this._running) {
+        return
+      }
+
       let responded = 0
 
       try {
@@ -83,16 +92,21 @@ class RoutingTable {
 
             try {
               timeoutController = new TimeoutController(this._pingTimeout)
+
               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.kb.remove(oldContact.id)
+              if (this._running) {
+                // only evict peers if we are still running, otherwise we evict when dialing is
+                // cancelled due to shutdown in progress
+                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) {
                 timeoutController.clear()
@@ -101,7 +115,7 @@ class RoutingTable {
           })
         )
 
-        if (responded < oldContacts.length) {
+        if (this._running && responded < oldContacts.length) {
           this._log(`adding new contact ${newContact.peer}`)
           this.kb.add(newContact)
         }

src/routing-table/refresh.js

@@ -218,10 +218,6 @@ class RoutingTableRefresh {
    * and the current peer
    */
   _maxCommonPrefix () {
-    if (!this._routingTable.kb.localNodeId) {
-      return 0
-    }
-
     // xor our KadId with every KadId in the k-bucket tree,
     // return the longest id prefix that is the same
     let prefixLength = 0
@@ -256,6 +252,10 @@ class RoutingTableRefresh {
    * Yields the common prefix length of every peer in the table
    */
   * _prefixLengths () {
+    if (!this._routingTable.kb) {
+      return
+    }
+
     for (const { id } of this._routingTable.kb.toIterable()) {
       const distance = uint8ArrayXor(this._routingTable.kb.localNodeId, id)
       let leadingZeros = 0

test/kad-dht.spec.js

@@ -766,11 +766,10 @@ describe('KadDHT', () => {
         tdht.connect(dhts[2], dhts[3])
       ])
 
-      const stub = sinon.stub(dhts[0]._lan._routingTable, 'closestPeers').returns([])
+      dhts[0]._lan.findPeer = sinon.stub().returns([])
+      dhts[0]._wan.findPeer = sinon.stub().returns([])
 
       await expect(drain(dhts[0].findPeer(ids[3]))).to.eventually.be.rejected().property('code', 'ERR_LOOKUP_FAILED')
-
-      stub.restore()
     })
   })
 })

test/routing-table.spec.js

@@ -26,6 +26,11 @@ describe('Routing Table', () => {
       peerId: lipbp2p.peerId,
       dialer: lipbp2p
     })
+    await table.start()
+  })
+
+  afterEach(async () => {
+    await table.stop()
   })
 
   it('add', async function () {
@@ -90,7 +95,8 @@ describe('Routing Table', () => {
     table._pingQueue = {
       add: (f) => {
         fn = f
-      }
+      },
+      clear: () => {}
     }
 
     const peerIds = [
@@ -135,7 +141,8 @@ describe('Routing Table', () => {
     table._pingQueue = {
       add: (f) => {
         fn = f
-      }
+      },
+      clear: () => {}
     }
 
     const peerIds = [