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

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,16 @@ export class KadDHT extends EventEmitter<PeerDiscoveryEvents> implements DHT {
       protocol: this.protocol,
       lan: this.lan
     })
+
+    // all queries should wait for the intial query-self query to run so we have
+    // some peers and don't force consumers to use arbitrary timeouts
+    const initialQuerySelfHasRun = pDefer<any>()
+
     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 +174,8 @@ export class KadDHT extends EventEmitter<PeerDiscoveryEvents> implements DHT {
     this.querySelf = new QuerySelf(components, {
       peerRouting: this.peerRouting,
       interval: querySelfInterval,
-      lan: this.lan
+      lan: this.lan,
+      initialQuerySelfHasRun
     })
 
     // handle peers being discovered during processing of DHT messages

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

@@ -8,13 +8,15 @@ 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'
 
 export interface QuerySelfInit {
   lan: boolean
   peerRouting: PeerRouting
   count?: number
   interval?: number
   queryTimeout?: number
+  initialQuerySelfHasRun: DeferredPromise<void>
 }
 
 /**
@@ -30,6 +32,7 @@ export class QuerySelf implements Startable {
   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
@@ -41,6 +44,7 @@ export class QuerySelf implements Startable {
     this.count = count ?? K
     this.interval = interval ?? QUERY_SELF_INTERVAL
     this.queryTimeout = queryTimeout ?? QUERY_SELF_TIMEOUT
+    this.initialQuerySelfHasRun = init.initialQuerySelfHasRun
   }
 
   isStarted (): boolean {
@@ -83,13 +87,19 @@ export class QuerySelf implements Startable {
       try {
         const found = await pipe(
           this.peerRouting.getClosestPeers(this.components.peerId.toBytes(), {
-            signal
+            signal,
+            isSelfQuery: true
           }),
           (source) => take(source, this.count),
           async (source) => await length(source)
         )
 
         this.log('query ran successfully - found %d peers', found)
+
+        if (this.initialQuerySelfHasRun != null) {
+          this.initialQuerySelfHasRun.resolve()
+          this.initialQuerySelfHasRun = undefined
+        }
       } catch (err: any) {
         this.log('query error', err)
       } finally {

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)