fix: re-enable ensuring queries run along disjoint paths (#371)

During the queries, the query paths might overlap and the same nodes will be queried multiple times.

There is no need to query a node if it has been queried during the execution of another query path, so the peersSeen set should be shared among different query paths belonging to the same query.

This also protects the network in that you have to control more nodes to affect query paths of other nodes.

Co-authored-by: Alex Potsides <alex@achingbrain.net>

package.json

@@ -150,6 +150,7 @@
     "@libp2p/interface-registrar": "^2.0.3",
     "@libp2p/interfaces": "^3.0.3",
     "@libp2p/logger": "^2.0.1",
+    "@libp2p/peer-collections": "^2.2.0",
     "@libp2p/peer-id": "^1.1.15",
     "@libp2p/record": "^2.0.2",
     "@libp2p/topology": "^3.0.0",

src/query/manager.ts

@@ -14,6 +14,7 @@ import type { Startable } from '@libp2p/interfaces/startable'
 import type { QueryFunc } from './types.js'
 import type { QueryOptions } from '@libp2p/interface-dht'
 import { Components, Initializable } from '@libp2p/components'
+import { PeerSet } from '@libp2p/peer-collections'
 
 const METRIC_RUNNING_QUERIES = 'running-queries'
 
@@ -139,6 +140,9 @@ export class QueryManager implements Startable, Initializable {
         return
       }
 
+      // make sure we don't get trapped in a loop
+      const peersSeen = new PeerSet()
+
       // Create query paths from the starting peers
       const paths = peersToQuery.map((peer, index) => {
         return queryPath({
@@ -152,7 +156,8 @@ export class QueryManager implements Startable, Initializable {
           alpha: this.alpha,
           cleanUp,
           queryFuncTimeout: options.queryFuncTimeout,
-          log
+          log,
+          peersSeen
         })
       })
 

src/query/query-path.ts

@@ -13,6 +13,7 @@ import type { CleanUpEvents } from './manager.js'
 import type { Logger } from '@libp2p/logger'
 import type { QueryFunc } from '../query/types.js'
 import type { QueryEvent } from '@libp2p/interface-dht'
+import type { PeerSet } from '@libp2p/peer-collections'
 
 const MAX_XOR = BigInt('0xFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFF')
 
@@ -71,14 +72,19 @@ export interface QueryPathOptions {
    * Query log
    */
   log: Logger
+
+  /**
+   * Set of peers seen by this and other paths
+   */
+  peersSeen: PeerSet
 }
 
 /**
  * Walks a path through the DHT, calling the passed query function for
  * every peer encountered that we have not seen before
  */
 export async function * queryPath (options: QueryPathOptions) {
-  const { key, startingPeer, ourPeerId, signal, query, alpha, pathIndex, numPaths, cleanUp, queryFuncTimeout, log } = options
+  const { key, startingPeer, ourPeerId, signal, query, alpha, pathIndex, numPaths, cleanUp, queryFuncTimeout, log, peersSeen } = options
   // Only ALPHA node/value lookups are allowed at any given time for each process
   // https://github.com/libp2p/specs/tree/master/kad-dht#alpha-concurrency-parameter-%CE%B1
   const queue = new Queue({
@@ -88,9 +94,6 @@ export async function * queryPath (options: QueryPathOptions) {
   // perform lookups on kadId, not the actual value
   const kadId = await convertBuffer(key)
 
-  // make sure we don't get trapped in a loop
-  const peersSeen = new Set()
-
   /**
    * Adds the passed peer to the query queue if it's not us and no
    * other path has passed through this peer
@@ -100,7 +103,7 @@ export async function * queryPath (options: QueryPathOptions) {
       return
     }
 
-    peersSeen.add(peer.toString())
+    peersSeen.add(peer)
 
     const peerXor = BigInt('0x' + toString(xor(peerKadId, kadId), 'base16'))
 
@@ -130,7 +133,7 @@ export async function * queryPath (options: QueryPathOptions) {
           // if there are closer peers and the query has not completed, continue the query
           if (event.name === 'PEER_RESPONSE') {
             for (const closerPeer of event.closer) {
-              if (peersSeen.has(closerPeer.id.toString())) { // eslint-disable-line max-depth
+              if (peersSeen.has(closerPeer.id)) { // eslint-disable-line max-depth
                 log('already seen %p in query', closerPeer.id)
                 continue
               }

test/query.spec.ts

@@ -427,6 +427,38 @@ describe('QueryManager', () => {
     await manager.stop()
   })
 
+  it('should stop when passing through the same node twice', async () => {
+    const manager = new QueryManager({ disjointPaths: 20, alpha: 1 })
+    manager.init(new Components({
+      peerId: ourPeerId
+    }))
+    await manager.start()
+
+    const topology = createTopology({
+      6: { closerPeers: [2] },
+      5: { closerPeers: [4] },
+      4: { closerPeers: [3] },
+      3: { closerPeers: [2] },
+      2: { closerPeers: [1] },
+      1: { closerPeers: [0] },
+      0: { value: uint8ArrayFromString('hello world') }
+    })
+
+    const results = await all(manager.run(key, [peers[6], peers[5]], createQueryFunction(topology)))
+    const traversedPeers = results
+      .map(event => {
+        if (event.type !== EventTypes.PEER_RESPONSE && event.type !== EventTypes.VALUE) {
+          throw new Error(`Unexpected query event type ${event.type}`)
+        }
+
+        return event.from
+      })
+
+    expect(traversedPeers).lengthOf(7)
+
+    await manager.stop()
+  })
+
   it('only closerPeers', async () => {
     const manager = new QueryManager({ disjointPaths: 1, alpha: 1 })
     manager.init(new Components({