fix: only choose query peers after initial self-query has run (#471)

Previously we chose peers before the self-query ran which meant there
was a reasonable chance we chose 0 peers.

Instead, wait for the query to run, then choose peers to query.

Author: achingbrain

src/content-fetching/index.ts

@@ -15,13 +15,12 @@ import {
   valueEvent,
   queryErrorEvent
 } from '../query/events.js'
-import { createPutRecord, convertBuffer, bufferToRecordKey } from '../utils.js'
+import { createPutRecord, bufferToRecordKey } from '../utils.js'
 import type { KadDHTComponents, Validators, Selectors, ValueEvent, QueryOptions, QueryEvent } from '../index.js'
 import type { Network } from '../network.js'
 import type { PeerRouting } from '../peer-routing/index.js'
 import type { QueryManager } from '../query/manager.js'
 import type { QueryFunc } from '../query/types.js'
-import type { RoutingTable } from '../routing-table/index.js'
 import type { AbortOptions } from '@libp2p/interfaces'
 import type { Logger } from '@libp2p/logger'
 
@@ -30,7 +29,6 @@ export interface ContentFetchingInit {
   selectors: Selectors
   peerRouting: PeerRouting
   queryManager: QueryManager
-  routingTable: RoutingTable
   network: Network
   lan: boolean
 }
@@ -42,19 +40,17 @@ export class ContentFetching {
   private readonly selectors: Selectors
   private readonly peerRouting: PeerRouting
   private readonly queryManager: QueryManager
-  private readonly routingTable: RoutingTable
   private readonly network: Network
 
   constructor (components: KadDHTComponents, init: ContentFetchingInit) {
-    const { validators, selectors, peerRouting, queryManager, routingTable, network, lan } = init
+    const { validators, selectors, peerRouting, queryManager, network, lan } = init
 
     this.components = components
     this.log = logger(`libp2p:kad-dht:${lan ? 'lan' : 'wan'}:content-fetching`)
     this.validators = validators
     this.selectors = selectors
     this.peerRouting = peerRouting
     this.queryManager = queryManager
-    this.routingTable = routingTable
     this.network = network
   }
 
@@ -249,11 +245,6 @@ export class ContentFetching {
       this.log('error getting local value for %b', key, err)
     }
 
-    const id = await convertBuffer(key)
-    const rtp = this.routingTable.closestPeers(id)
-
-    this.log('found %d peers in routing table', rtp.length)
-
     const self = this // eslint-disable-line @typescript-eslint/no-this-alias
 
     const getValueQuery: QueryFunc = async function * ({ peer, signal }) {
@@ -267,6 +258,6 @@ export class ContentFetching {
     }
 
     // we have peers, lets send the actual query to them
-    yield * this.queryManager.run(key, rtp, getValueQuery, options)
+    yield * this.queryManager.run(key, getValueQuery, options)
   }
 }

src/content-routing/index.ts

@@ -9,7 +9,6 @@ import {
   peerResponseEvent,
   providerEvent
 } from '../query/events.js'
-import { convertBuffer } from '../utils.js'
 import type { KadDHTComponents, PeerResponseEvent, ProviderEvent, QueryEvent, QueryOptions } from '../index.js'
 import type { Network } from '../network.js'
 import type { PeerRouting } from '../peer-routing/index.js'
@@ -126,7 +125,6 @@ export class ContentRouting {
   async * findProviders (key: CID, options: QueryOptions): AsyncGenerator<PeerResponseEvent | ProviderEvent | QueryEvent> {
     const toFind = this.routingTable.kBucketSize
     const target = key.multihash.bytes
-    const id = await convertBuffer(target)
     const self = this // eslint-disable-line @typescript-eslint/no-this-alias
 
     this.log('findProviders %c', key)
@@ -175,7 +173,7 @@ export class ContentRouting {
 
     const providers = new Set(provs.map(p => p.toString()))
 
-    for await (const event of this.queryManager.run(target, this.routingTable.closestPeers(id), findProvidersQuery, options)) {
+    for await (const event of this.queryManager.run(target, findProvidersQuery, options)) {
       yield event
 
       if (event.name === 'PEER_RESPONSE') {

src/kad-dht.ts

@@ -131,7 +131,8 @@ export class DefaultKadDHT extends EventEmitter<PeerDiscoveryEvents> implements
       // 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,
-      initialQuerySelfHasRun
+      initialQuerySelfHasRun,
+      routingTable: this.routingTable
     })
 
     // DHT components
@@ -147,7 +148,6 @@ export class DefaultKadDHT extends EventEmitter<PeerDiscoveryEvents> implements
       selectors: this.selectors,
       peerRouting: this.peerRouting,
       queryManager: this.queryManager,
-      routingTable: this.routingTable,
       network: this.network,
       lan: this.lan
     })

src/peer-routing/index.ts

@@ -148,34 +148,6 @@ export class PeerRouting {
       return
     }
 
-    const key = await utils.convertPeerId(id)
-    const peers = this.routingTable.closestPeers(key)
-
-    // sanity check
-    const match = peers.find((p) => p.equals(id))
-
-    if (match != null) {
-      try {
-        const peer = await this.components.peerStore.get(id)
-
-        this.log('found in peerStore')
-        yield finalPeerEvent({
-          from: this.components.peerId,
-          peer: {
-            id: peer.id,
-            multiaddrs: peer.addresses.map((address) => address.multiaddr),
-            protocols: []
-          }
-        })
-
-        return
-      } catch (err: any) {
-        if (err.code !== 'ERR_NOT_FOUND') {
-          throw err
-        }
-      }
-    }
-
     const self = this // eslint-disable-line @typescript-eslint/no-this-alias
 
     const findPeerQuery: QueryFunc = async function * ({ peer, signal }) {
@@ -197,7 +169,7 @@ export class PeerRouting {
 
     let foundPeer = false
 
-    for await (const event of this.queryManager.run(id.toBytes(), peers, findPeerQuery, options)) {
+    for await (const event of this.queryManager.run(id.toBytes(), findPeerQuery, options)) {
       if (event.name === 'FINAL_PEER') {
         foundPeer = true
       }
@@ -230,7 +202,7 @@ export class PeerRouting {
       yield * self.network.sendRequest(peer, request, { signal })
     }
 
-    for await (const event of this.queryManager.run(key, tablePeers, getCloserPeersQuery, options)) {
+    for await (const event of this.queryManager.run(key, getCloserPeersQuery, options)) {
       yield event
 
       if (event.name === 'PEER_RESPONSE') {

src/query/manager.ts

@@ -9,9 +9,11 @@ import { toString as uint8ArrayToString } from 'uint8arrays/to-string'
 import {
   ALPHA, K, DEFAULT_QUERY_TIMEOUT
 } from '../constants.js'
+import { convertBuffer } from '../utils.js'
 import { queryPath } from './query-path.js'
 import type { QueryFunc } from './types.js'
 import type { QueryEvent } from '../index.js'
+import type { RoutingTable } from '../routing-table/index.js'
 import type { Metric, Metrics } from '@libp2p/interface-metrics'
 import type { PeerId } from '@libp2p/interface-peer-id'
 import type { AbortOptions } from '@libp2p/interfaces'
@@ -27,6 +29,7 @@ export interface QueryManagerInit {
   disjointPaths?: number
   alpha?: number
   initialQuerySelfHasRun: DeferredPromise<void>
+  routingTable: RoutingTable
 }
 
 export interface QueryManagerComponents {
@@ -55,6 +58,7 @@ export class QueryManager implements Startable {
     queryTime: Metric
   }
 
+  private readonly routingTable: RoutingTable
   private initialQuerySelfHasRun?: DeferredPromise<void>
 
   constructor (components: QueryManagerComponents, init: QueryManagerInit) {
@@ -67,6 +71,7 @@ export class QueryManager implements Startable {
     this.lan = lan
     this.queries = 0
     this.initialQuerySelfHasRun = init.initialQuerySelfHasRun
+    this.routingTable = init.routingTable
 
     // allow us to stop queries on shut down
     this.shutDownController = new AbortController()
@@ -105,7 +110,7 @@ export class QueryManager implements Startable {
     this.shutDownController.abort()
   }
 
-  async * run (key: Uint8Array, peers: PeerId[], queryFunc: QueryFunc, options: QueryOptions = {}): AsyncGenerator<QueryEvent> {
+  async * run (key: Uint8Array, queryFunc: QueryFunc, options: QueryOptions = {}): AsyncGenerator<QueryEvent> {
     if (!this.running) {
       throw new Error('QueryManager not started')
     }
@@ -138,7 +143,6 @@ export class QueryManager implements Startable {
     const log = logger(`libp2p:kad-dht:${this.lan ? 'lan' : 'wan'}:query:` + uint8ArrayToString(key, 'base58btc'))
 
     // query a subset of peers up to `kBucketSize / 2` in length
-    const peersToQuery = peers.slice(0, Math.min(this.disjointPaths, peers.length))
     const startTime = Date.now()
     const cleanUp = new EventEmitter<CleanUpEvents>()
 
@@ -162,6 +166,10 @@ export class QueryManager implements Startable {
       this.queries++
       this.metrics?.runningQueries.update(this.queries)
 
+      const id = await convertBuffer(key)
+      const peers = this.routingTable.closestPeers(id)
+      const peersToQuery = peers.slice(0, Math.min(this.disjointPaths, peers.length))
+
       if (peers.length === 0) {
         log.error('Running query with no peers')
         return