fix: add default query timeouts (#266)

If an `AbortSignal` is not passed to the query manager, create a
default timeout to prevent queries from running forever.

Also add configurable default timeouts for self-query and table refresh.

Author: achingbrain

src/constants.js

@@ -39,5 +39,14 @@ exports.ALPHA = 3
 // How often we look for our closest DHT neighbours
 exports.QUERY_SELF_INTERVAL = Number(5 * minute)
 
+// How long to look for our closest DHT neighbours for
+exports.QUERY_SELF_TIMEOUT = Number(30 * second)
+
 // How often we try to find new peers
 exports.TABLE_REFRESH_INTERVAL = Number(5 * minute)
+
+// How how long to look for new peers for
+exports.TABLE_REFRESH_QUERY_TIMEOUT = Number(30 * second)
+
+// When a timeout is not specified, run a query for this long
+exports.DEFAULT_QUERY_TIMEOUT = Number(30 * second)

src/kad-dht.js

@@ -363,7 +363,6 @@ class KadDHT extends EventEmitter {
       this._queryManager.start(),
       this._network.start(),
       this._routingTable.start(),
-      this._routingTableRefresh.start(),
       this._topologyListener.start(),
       this._querySelf.start()
     ])
@@ -372,6 +371,7 @@ class KadDHT extends EventEmitter {
       this._bootstrapPeers.map(peerData => this._routingTable.add(peerData.id))
     )
 
+    await this._routingTableRefresh.start()
     await this.refreshRoutingTable()
   }
 

src/query-self.js

@@ -3,8 +3,10 @@
 const { EventEmitter } = require('events')
 const take = require('it-take')
 const length = require('it-length')
-const { QUERY_SELF_INTERVAL, K } = require('./constants')
+const { QUERY_SELF_INTERVAL, QUERY_SELF_TIMEOUT, K } = require('./constants')
 const utils = require('./utils')
+const { TimeoutController } = require('timeout-abort-controller')
+const { anySignal } = require('any-signal')
 
 /**
  * Receives notifications of new peers joining the network that support the DHT protocol
@@ -18,17 +20,19 @@ class QuerySelf extends EventEmitter {
    * @param {import('./peer-routing').PeerRouting} params.peerRouting
    * @param {number} [params.count] - how many peers to find
    * @param {number} [params.interval] - how often to find them
+   * @param {number} [params.queryTimeout] - how long to let queries run
    * @param {boolean} params.lan
    */
-  constructor ({ peerId, peerRouting, lan, count = K, interval = QUERY_SELF_INTERVAL }) {
+  constructor ({ peerId, peerRouting, lan, count = K, interval = QUERY_SELF_INTERVAL, queryTimeout = QUERY_SELF_TIMEOUT }) {
     super()
 
     this._log = utils.logger(`libp2p:kad-dht:${lan ? 'lan' : 'wan'}:query-self`)
     this._running = false
     this._peerId = peerId
     this._peerRouting = peerRouting
-    this._count = count
-    this._interval = interval
+    this._count = count || K
+    this._interval = interval || QUERY_SELF_INTERVAL
+    this._queryTimeout = queryTimeout || QUERY_SELF_TIMEOUT
   }
 
   /**
@@ -59,18 +63,20 @@ class QuerySelf extends EventEmitter {
   }
 
   async _querySelf () {
+    const timeoutController = new TimeoutController(this._queryTimeout)
+
     try {
       this._controller = new AbortController()
-
       const found = await length(await take(this._peerRouting.getClosestPeers(this._peerId.toBytes(), {
-        signal: this._controller.signal
+        signal: anySignal([this._controller.signal, timeoutController.signal])
       }), this._count))
 
       this._log('query ran successfully - found %d peers', found)
     } catch (/** @type {any} */ err) {
       this._log('query error', err)
     } finally {
       this._timeoutId = setTimeout(this._querySelf.bind(this), this._interval)
+      timeoutController.clear()
     }
   }
 }

src/query/manager.js

@@ -1,9 +1,10 @@
 'use strict'
 
 const { AbortController } = require('native-abort-controller')
+const { TimeoutController } = require('timeout-abort-controller')
 const { anySignal } = require('any-signal')
 const {
-  ALPHA, K
+  ALPHA, K, DEFAULT_QUERY_TIMEOUT
 } = require('../constants')
 const { toString: uint8ArrayToString } = require('uint8arrays/to-string')
 const { logger } = require('../utils')
@@ -78,6 +79,14 @@ class QueryManager {
       throw new Error('QueryManager not started')
     }
 
+    let timeoutController
+
+    if (!options.signal) {
+      // don't let queries run forever
+      timeoutController = new TimeoutController(DEFAULT_QUERY_TIMEOUT)
+      options.signal = timeoutController.signal
+    }
+
     // allow us to stop queries on shut down
     const abortController = new AbortController()
     this._controllers.add(abortController)
@@ -139,6 +148,11 @@ class QueryManager {
       }
     } finally {
       this._controllers.delete(abortController)
+
+      if (timeoutController) {
+        timeoutController.clear()
+      }
+
       cleanUp.emit('cleanup')
       log(`query:done in ${Date.now() - (startTime || 0)}ms`)
     }

src/routing-table/refresh.js

@@ -8,7 +8,7 @@ const PeerId = require('peer-id')
 const utils = require('../utils')
 const length = require('it-length')
 const { TimeoutController } = require('timeout-abort-controller')
-const { TABLE_REFRESH_INTERVAL } = require('../constants')
+const { TABLE_REFRESH_INTERVAL, TABLE_REFRESH_QUERY_TIMEOUT } = require('../constants')
 
 /**
  * @typedef {import('./types').KBucketPeer} KBucketPeer
@@ -32,12 +32,14 @@ class RoutingTableRefresh {
    * @param {import('./').RoutingTable} params.routingTable
    * @param {boolean} params.lan
    * @param {number} [params.refreshInterval]
+   * @param {number} [params.refreshQueryTimeout]
    */
-  constructor ({ peerRouting, routingTable, refreshInterval, lan }) {
+  constructor ({ peerRouting, routingTable, refreshInterval, refreshQueryTimeout, lan }) {
     this._log = utils.logger(`libp2p:kad-dht:${lan ? 'lan' : 'wan'}:routing-table:refresh`)
     this._peerRouting = peerRouting
     this._routingTable = routingTable
     this._refreshInterval = refreshInterval || TABLE_REFRESH_INTERVAL
+    this._refreshQueryTimeout = refreshQueryTimeout || TABLE_REFRESH_QUERY_TIMEOUT
 
     /** @type {Date[]} */
     this.commonPrefixLengthRefreshedAt = []
@@ -135,7 +137,7 @@ class RoutingTableRefresh {
 
     this._log('starting refreshing cpl %s with key %p (routing table size was %s)', cpl, peerId, this._routingTable.kb.count())
 
-    const controller = new TimeoutController(60000)
+    const controller = new TimeoutController(this._refreshQueryTimeout)
 
     try {
       const peers = await length(this._peerRouting.getClosestPeers(peerId.toBytes(), { signal: controller.signal }))