fix: wait for self-query to have run before running queries (#457)

In order to run a query we need DHT peers. The most common way to
find these is to run a self-query which happens after startup.

There's no way for the users to know when this has happened so they
end up putting arbitrary waits into the code so their queries have
a better chance of succeeding.

Instead just delay all queries until the self-query has run and
found some peers, then continue with the queries.

Author: achingbrain

src/constants.ts

@@ -41,8 +41,11 @@ export const ALPHA = 3
 // How often we look for our closest DHT neighbours
 export const QUERY_SELF_INTERVAL = Number(5 * minute)
 
+// How often we look for the first set of our closest DHT neighbours
+export const QUERY_SELF_INITIAL_INTERVAL = Number(Number(second))
+
 // How long to look for our closest DHT neighbours for
-export const QUERY_SELF_TIMEOUT = Number(30 * second)
+export const QUERY_SELF_TIMEOUT = Number(5 * second)
 
 // How often we try to find new peers
 export const TABLE_REFRESH_INTERVAL = Number(5 * minute)

src/index.ts

@@ -45,6 +45,21 @@ export interface KadDHTInit {
    */
   querySelfInterval?: number
 
+  /**
+   * During startup we run the self-query at a shorter interval to ensure
+   * the containing node can respond to queries quickly. Set this interval
+   * here in ms (default: 1000)
+   */
+  initialQuerySelfInterval?: number
+
+  /**
+   * After startup by default all queries will be paused until the initial
+   * self-query has run and there are some peers in the routing table.
+   *
+   * Pass true here to disable this behaviour. (default: false)
+   */
+  allowQueryWithZeroPeers?: boolean
+
   /**
    * A custom protocol prefix to use (default: '/ipfs')
    */

src/kad-dht.ts

@@ -25,6 +25,7 @@ import { validators as recordValidators } from '@libp2p/record/validators'
 import { selectors as recordSelectors } from '@libp2p/record/selectors'
 import { symbol } from '@libp2p/interface-peer-discovery'
 import { PROTOCOL_DHT, PROTOCOL_PREFIX, LAN_PREFIX } from './constants.js'
+import pDefer from 'p-defer'
 
 export const DEFAULT_MAX_INBOUND_STREAMS = 32
 export const DEFAULT_MAX_OUTBOUND_STREAMS = 64
@@ -117,10 +118,22 @@ export class KadDHT extends EventEmitter<PeerDiscoveryEvents> implements DHT {
       protocol: this.protocol,
       lan: this.lan
     })
+
+    // all queries should wait for the initial query-self query to run so we have
+    // some peers and don't force consumers to use arbitrary timeouts
+    const initialQuerySelfHasRun = pDefer<any>()
+
+    // if the user doesn't want to wait for query peers, resolve the initial
+    // self-query promise immediately
+    if (init.allowQueryWithZeroPeers === true) {
+      initialQuerySelfHasRun.resolve()
+    }
+
     this.queryManager = new QueryManager(components, {
       // Number of disjoint query paths to use - This is set to `kBucketSize/2` per the S/Kademlia paper
       disjointPaths: Math.ceil(this.kBucketSize / 2),
-      lan
+      lan,
+      initialQuerySelfHasRun
     })
 
     // DHT components
@@ -167,7 +180,10 @@ export class KadDHT extends EventEmitter<PeerDiscoveryEvents> implements DHT {
     this.querySelf = new QuerySelf(components, {
       peerRouting: this.peerRouting,
       interval: querySelfInterval,
-      lan: this.lan
+      initialInterval: init.initialQuerySelfInterval,
+      lan: this.lan,
+      initialQuerySelfHasRun,
+      routingTable: this.routingTable
     })
 
     // handle peers being discovered during processing of DHT messages
@@ -212,7 +228,7 @@ export class KadDHT extends EventEmitter<PeerDiscoveryEvents> implements DHT {
   }
 
   async onPeerConnect (peerData: PeerInfo): Promise<void> {
-    this.log('peer %p connected with protocols %s', peerData.id, peerData.protocols)
+    this.log('peer %p connected with protocols', peerData.id, peerData.protocols)
 
     if (this.lan) {
       peerData = removePublicAddresses(peerData)

src/peer-routing/index.ts

@@ -13,9 +13,9 @@ import { Libp2pRecord } from '@libp2p/record'
 import { logger } from '@libp2p/logger'
 import { keys } from '@libp2p/crypto'
 import { peerIdFromKeys } from '@libp2p/peer-id'
-import type { DHTRecord, DialingPeerEvent, FinalPeerEvent, QueryEvent, QueryOptions, Validators } from '@libp2p/interface-dht'
+import type { DHTRecord, DialingPeerEvent, FinalPeerEvent, QueryEvent, Validators } from '@libp2p/interface-dht'
 import type { RoutingTable } from '../routing-table/index.js'
-import type { QueryManager } from '../query/manager.js'
+import type { QueryManager, QueryOptions } from '../query/manager.js'
 import type { Network } from '../network.js'
 import type { Logger } from '@libp2p/logger'
 import type { AbortOptions } from '@libp2p/interfaces'

src/query-self.ts

@@ -1,20 +1,39 @@
 import { setMaxListeners } from 'events'
 import take from 'it-take'
 import length from 'it-length'
-import { QUERY_SELF_INTERVAL, QUERY_SELF_TIMEOUT, K } from './constants.js'
+import { QUERY_SELF_INTERVAL, QUERY_SELF_TIMEOUT, K, QUERY_SELF_INITIAL_INTERVAL } from './constants.js'
 import { anySignal } from 'any-signal'
 import { logger, Logger } from '@libp2p/logger'
 import type { PeerRouting } from './peer-routing/index.js'
 import type { Startable } from '@libp2p/interfaces/startable'
 import { pipe } from 'it-pipe'
 import type { KadDHTComponents } from './index.js'
+import type { DeferredPromise } from 'p-defer'
+import type { RoutingTable } from './routing-table/index.js'
 
 export interface QuerySelfInit {
   lan: boolean
   peerRouting: PeerRouting
+  routingTable: RoutingTable
   count?: number
   interval?: number
+  initialInterval?: number
   queryTimeout?: number
+  initialQuerySelfHasRun: DeferredPromise<void>
+}
+
+function debounce (func: () => void, wait: number): () => void {
+  let timeout: ReturnType<typeof setTimeout> | undefined
+
+  return function () {
+    const later = function (): void {
+      timeout = undefined
+      func()
+    }
+
+    clearTimeout(timeout)
+    timeout = setTimeout(later, wait)
+  }
 }
 
 /**
@@ -24,40 +43,51 @@ export class QuerySelf implements Startable {
   private readonly log: Logger
   private readonly components: KadDHTComponents
   private readonly peerRouting: PeerRouting
+  private readonly routingTable: RoutingTable
   private readonly count: number
   private readonly interval: number
+  private readonly initialInterval: number
   private readonly queryTimeout: number
+  private started: boolean
   private running: boolean
   private timeoutId?: NodeJS.Timer
   private controller?: AbortController
+  private initialQuerySelfHasRun?: DeferredPromise<void>
 
   constructor (components: KadDHTComponents, init: QuerySelfInit) {
-    const { peerRouting, lan, count, interval, queryTimeout } = init
+    const { peerRouting, lan, count, interval, queryTimeout, routingTable } = init
 
     this.components = components
     this.log = logger(`libp2p:kad-dht:${lan ? 'lan' : 'wan'}:query-self`)
     this.running = false
+    this.started = false
     this.peerRouting = peerRouting
+    this.routingTable = routingTable
     this.count = count ?? K
     this.interval = interval ?? QUERY_SELF_INTERVAL
+    this.initialInterval = init.initialInterval ?? QUERY_SELF_INITIAL_INTERVAL
     this.queryTimeout = queryTimeout ?? QUERY_SELF_TIMEOUT
+    this.initialQuerySelfHasRun = init.initialQuerySelfHasRun
+
+    this.querySelf = debounce(this.querySelf.bind(this), 100)
   }
 
   isStarted (): boolean {
-    return this.running
+    return this.started
   }
 
   async start (): Promise<void> {
-    if (this.running) {
+    if (this.started) {
       return
     }
 
-    this.running = true
-    this._querySelf()
+    this.started = true
+    clearTimeout(this.timeoutId)
+    this.timeoutId = setTimeout(this.querySelf.bind(this), this.initialInterval)
   }
 
   async stop (): Promise<void> {
-    this.running = false
+    this.started = false
 
     if (this.timeoutId != null) {
       clearTimeout(this.timeoutId)
@@ -68,36 +98,76 @@ export class QuerySelf implements Startable {
     }
   }
 
-  _querySelf (): void {
-    Promise.resolve().then(async () => {
-      this.controller = new AbortController()
-      const signal = anySignal([this.controller.signal, AbortSignal.timeout(this.queryTimeout)])
+  querySelf (): void {
+    if (!this.started) {
+      this.log('skip self-query because we are not started')
+      return
+    }
 
-      // this controller will get used for lots of dial attempts so make sure we don't cause warnings to be logged
-      try {
-        if (setMaxListeners != null) {
-          setMaxListeners(Infinity, signal)
-        }
-      } catch {} // fails on node < 15.4
-
-      try {
-        const found = await pipe(
-          this.peerRouting.getClosestPeers(this.components.peerId.toBytes(), {
-            signal
-          }),
-          (source) => take(source, this.count),
-          async (source) => await length(source)
-        )
-
-        this.log('query ran successfully - found %d peers', found)
-      } catch (err: any) {
-        this.log('query error', err)
-      } finally {
-        this.timeoutId = setTimeout(this._querySelf.bind(this), this.interval)
-        signal.clear()
+    if (this.running) {
+      this.log('skip self-query because we are already running, will run again in %dms', this.interval)
+      return
+    }
+
+    if (this.routingTable.size === 0) {
+      let nextInterval = this.interval
+
+      if (this.initialQuerySelfHasRun != null) {
+        // if we've not yet run the first self query, shorten the interval until we try again
+        nextInterval = this.initialInterval
       }
-    }).catch(err => {
-      this.log('query error', err)
-    })
+
+      this.log('skip self-query because routing table is empty, will run again in %dms', nextInterval)
+      clearTimeout(this.timeoutId)
+      this.timeoutId = setTimeout(this.querySelf.bind(this), nextInterval)
+      return
+    }
+
+    this.running = true
+
+    Promise.resolve()
+      .then(async () => {
+        this.controller = new AbortController()
+        const signal = anySignal([this.controller.signal, AbortSignal.timeout(this.queryTimeout)])
+
+        // this controller will get used for lots of dial attempts so make sure we don't cause warnings to be logged
+        try {
+          if (setMaxListeners != null) {
+            setMaxListeners(Infinity, signal)
+          }
+        } catch {} // fails on node < 15.4
+
+        try {
+          this.log('run self-query, look for %d peers timing out after %dms', this.count, this.queryTimeout)
+
+          const found = await pipe(
+            this.peerRouting.getClosestPeers(this.components.peerId.toBytes(), {
+              signal,
+              isSelfQuery: true
+            }),
+            (source) => take(source, this.count),
+            async (source) => await length(source)
+          )
+
+          this.log('self-query ran successfully - found %d peers', found)
+
+          if (this.initialQuerySelfHasRun != null) {
+            this.initialQuerySelfHasRun.resolve()
+            this.initialQuerySelfHasRun = undefined
+          }
+        } catch (err: any) {
+          this.log.error('self-query error', err)
+        } finally {
+          signal.clear()
+        }
+      }).catch(err => {
+        this.log('self-query error', err)
+      }).finally(() => {
+        this.running = false
+
+        this.log('running self-query again in %dms', this.interval)
+        clearTimeout(this.timeoutId)
+        this.timeoutId = setTimeout(this.querySelf.bind(this), this.interval)
+      })
   }
 }

src/query/manager.ts

@@ -11,9 +11,12 @@ import { logger } from '@libp2p/logger'
 import type { PeerId } from '@libp2p/interface-peer-id'
 import type { Startable } from '@libp2p/interfaces/startable'
 import type { QueryFunc } from './types.js'
-import type { QueryEvent, QueryOptions } from '@libp2p/interface-dht'
+import type { QueryEvent } from '@libp2p/interface-dht'
 import { PeerSet } from '@libp2p/peer-collections'
 import type { Metric, Metrics } from '@libp2p/interface-metrics'
+import type { DeferredPromise } from 'p-defer'
+import type { AbortOptions } from '@libp2p/interfaces'
+import { AbortError } from '@libp2p/interfaces/errors'
 
 export interface CleanUpEvents {
   'cleanup': CustomEvent
@@ -23,13 +26,19 @@ export interface QueryManagerInit {
   lan?: boolean
   disjointPaths?: number
   alpha?: number
+  initialQuerySelfHasRun: DeferredPromise<void>
 }
 
 export interface QueryManagerComponents {
   peerId: PeerId
   metrics?: Metrics
 }
 
+export interface QueryOptions extends AbortOptions {
+  queryFuncTimeout?: number
+  isSelfQuery?: boolean
+}
+
 /**
  * Keeps track of all running queries
  */
@@ -46,6 +55,8 @@ export class QueryManager implements Startable {
     queryTime: Metric
   }
 
+  private initialQuerySelfHasRun?: DeferredPromise<void>
+
   constructor (components: QueryManagerComponents, init: QueryManagerInit) {
     const { lan = false, disjointPaths = K, alpha = ALPHA } = init
 
@@ -55,6 +66,7 @@ export class QueryManager implements Startable {
     this.alpha = alpha ?? ALPHA
     this.lan = lan
     this.queries = 0
+    this.initialQuerySelfHasRun = init.initialQuerySelfHasRun
 
     // allow us to stop queries on shut down
     this.shutDownController = new AbortController()
@@ -131,6 +143,21 @@ export class QueryManager implements Startable {
     const cleanUp = new EventEmitter<CleanUpEvents>()
 
     try {
+      if (options.isSelfQuery !== true && this.initialQuerySelfHasRun != null) {
+        log('waiting for initial query-self query before continuing')
+
+        await Promise.race([
+          new Promise((resolve, reject) => {
+            signal.addEventListener('abort', () => {
+              reject(new AbortError('Query was aborted before self-query ran'))
+            })
+          }),
+          this.initialQuerySelfHasRun.promise
+        ])
+
+        this.initialQuerySelfHasRun = undefined
+      }
+
       log('query:start')
       this.queries++
       this.metrics?.runningQueries.update(this.queries)