fix: wait for self-query to have run before running queries

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)