fix: ensure greater spec compliance (#3105)

- Increase alpha to 10
- Split starting peers into multiple disjoint buckets
- Report disjoint path info on events
- Abort slow subqueries at the network level instead of query and network
- Use a filter to separate disjoint paths
- Only add closer peers to our list of closer peers if they are actually closer

Author: achingbrain

packages/kad-dht/package.json

@@ -77,6 +77,7 @@
     "it-parallel": "^3.0.8",
     "it-pipe": "^3.0.1",
     "it-protobuf-stream": "^2.0.1",
+    "it-pushable": "^3.2.3",
     "it-take": "^3.0.6",
     "mortice": "^3.0.6",
     "multiformats": "^13.3.1",

packages/kad-dht/src/constants.ts

@@ -36,7 +36,7 @@ export const GET_MANY_RECORD_COUNT = 16
 export const K = 20
 
 // Alpha is the concurrency for asynchronous requests
-export const ALPHA = 3
+export const ALPHA = 10
 
 // How often we look for our closest DHT neighbors
 export const QUERY_SELF_INTERVAL = 5 * minute

packages/kad-dht/src/content-fetching/index.ts

@@ -129,7 +129,7 @@ export class ContentFetching {
       }
 
       if (!sentCorrection) {
-        yield queryErrorEvent({ from, error: new QueryError('Value not put correctly') }, options)
+        throw new QueryError('Could not send correction')
       }
 
       this.log.error('Failed error correcting entry')
@@ -182,7 +182,11 @@ export class ContentFetching {
             }
 
             if (!(putEvent.record != null && uint8ArrayEquals(putEvent.record.value, Libp2pRecord.deserialize(record).value))) {
-              events.push(queryErrorEvent({ from: event.peer.id, error: new QueryError('Value not put correctly') }, options))
+              events.push(queryErrorEvent({
+                from: event.peer.id,
+                error: new QueryError('Value not put correctly'),
+                path: putEvent.path
+              }, options))
             }
           }
 
@@ -212,6 +216,7 @@ export class ContentFetching {
     for await (const event of this.getMany(key, options)) {
       if (event.name === 'VALUE') {
         vals.push(event)
+        continue
       }
 
       yield event
@@ -242,7 +247,12 @@ export class ContentFetching {
 
     yield * this.sendCorrectionRecord(key, vals, best, {
       ...options,
-      path: -1
+      path: {
+        index: -1,
+        queued: 0,
+        running: 0,
+        total: 0
+      }
     })
 
     yield vals[i]
@@ -259,7 +269,13 @@ export class ContentFetching {
 
       yield valueEvent({
         value: localRec.value,
-        from: this.components.peerId
+        from: this.components.peerId,
+        path: {
+          index: -1,
+          running: 0,
+          queued: 0,
+          total: 0
+        }
       }, options)
     } catch (err: any) {
       this.log('error getting local value for %b', key, err)
@@ -268,15 +284,19 @@ export class ContentFetching {
     const self = this // eslint-disable-line @typescript-eslint/no-this-alias
 
     const getValueQuery: QueryFunc = async function * ({ peer, signal, path }) {
-      for await (const event of self.peerRouting.getValueOrPeers(peer, key, {
+      for await (const event of self.peerRouting.getValueOrPeers(peer.id, key, {
         ...options,
         signal,
         path
       })) {
         yield event
 
         if (event.name === 'PEER_RESPONSE' && (event.record != null)) {
-          yield valueEvent({ from: peer, value: event.record.value }, options)
+          yield valueEvent({
+            from: peer.id,
+            value: event.record.value,
+            path
+          }, options)
         }
       }
     }

packages/kad-dht/src/content-routing/index.ts

@@ -1,7 +1,6 @@
 import { PeerSet } from '@libp2p/peer-collections'
-import map from 'it-map'
-import parallel from 'it-parallel'
-import { pipe } from 'it-pipe'
+import { Queue } from '@libp2p/utils/queue'
+import { pushable } from 'it-pushable'
 import { ALPHA } from '../constants.js'
 import { MessageType } from '../message/dht.js'
 import { toPbPeerInfo } from '../message/utils.js'
@@ -10,7 +9,7 @@ import {
   peerResponseEvent,
   providerEvent
 } from '../query/events.js'
-import type { KadDHTComponents, PeerResponseEvent, ProviderEvent, QueryEvent } from '../index.js'
+import type { FinalPeerEvent, KadDHTComponents, PeerResponseEvent, ProviderEvent, QueryEvent } from '../index.js'
 import type { Message } from '../message/dht.js'
 import type { Network } from '../network.js'
 import type { PeerRouting } from '../peer-routing/index.js'
@@ -98,54 +97,76 @@ export class ContentRouting {
     }
 
     let sent = 0
+    const self = this
+
+    async function * publishProviderRecord (event: FinalPeerEvent): AsyncGenerator<QueryEvent, void, undefined> {
+      try {
+        self.log('sending provider record for %s to %p', key, event.peer.id)
+
+        for await (const addProviderEvent of self.network.sendMessage(event.peer.id, msg, {
+          ...options,
+          path: event.path
+        })) {
+          if (addProviderEvent.name === 'PEER_RESPONSE') {
+            self.log('sent provider record for %s to %p', key, event.peer.id)
+            sent++
+          }
 
-    const maybeNotifyPeer = (event: QueryEvent) => {
-      return async () => {
-        if (event.name !== 'FINAL_PEER') {
-          return [event]
+          yield addProviderEvent
         }
+      } catch (err: any) {
+        self.log.error('error sending provide record to peer %p', event.peer.id, err)
+        yield queryErrorEvent({
+          from: event.peer.id,
+          error: err,
+          path: event.path
+        }, options)
+      }
+    }
 
-        const events = []
+    const events = pushable<QueryEvent>({
+      objectMode: true
+    })
 
-        this.log('putProvider %s to %p', key, event.peer.id)
+    const queue = new Queue({
+      concurrency: ALPHA
+    })
+    queue.addEventListener('idle', () => {
+      events.end()
+    })
+    queue.addEventListener('error', (err) => {
+      this.log.error('error publishing provider record to peer - %e', err)
+    })
 
-        try {
-          this.log('sending provider record for %s to %p', key, event.peer.id)
-
-          for await (const sendEvent of this.network.sendMessage(event.peer.id, msg, {
-            ...options,
-            path: event.path ?? -1
-          })) {
-            if (sendEvent.name === 'PEER_RESPONSE') {
-              this.log('sent provider record for %s to %p', key, event.peer.id)
-              sent++
-            }
-
-            events.push(sendEvent)
-          }
-        } catch (err: any) {
-          this.log.error('error sending provide record to peer %p', event.peer.id, err)
-          events.push(queryErrorEvent({ from: event.peer.id, error: err }, options))
-        }
+    queue.add(async () => {
+      const finalPeerEvents: FinalPeerEvent[] = []
 
-        return events
-      }
-    }
+      for await (const event of this.peerRouting.getClosestPeers(target, options)) {
+        events.push(event)
 
-    // Notify closest peers
-    yield * pipe(
-      this.peerRouting.getClosestPeers(target, options),
-      (source) => map(source, (event) => maybeNotifyPeer(event)),
-      (source) => parallel(source, {
-        ordered: false,
-        concurrency: ALPHA
-      }),
-      async function * (source) {
-        for await (const events of source) {
-          yield * events
+        if (event.name !== 'FINAL_PEER') {
+          continue
         }
+
+        finalPeerEvents.push(event)
       }
-    )
+
+      finalPeerEvents.forEach(event => {
+        queue.add(async () => {
+          for await (const notifyEvent of publishProviderRecord(event)) {
+            events.push(notifyEvent)
+          }
+        })
+          .catch(err => {
+            this.log.error('error publishing provider record to peer - %e', err)
+          })
+      })
+    })
+      .catch(err => {
+        events.end(err)
+      })
+
+    yield * events
 
     this.log('sent provider records to %d peers', sent)
   }
@@ -184,8 +205,27 @@ export class ContentRouting {
         }
       }
 
-      yield peerResponseEvent({ from: this.components.peerId, messageType: MessageType.GET_PROVIDERS, providers, path: -1 }, options)
-      yield providerEvent({ from: this.components.peerId, providers }, options)
+      yield peerResponseEvent({
+        from: this.components.peerId,
+        messageType: MessageType.GET_PROVIDERS,
+        providers,
+        path: {
+          index: -1,
+          queued: 0,
+          running: 0,
+          total: 0
+        }
+      }, options)
+      yield providerEvent({
+        from: this.components.peerId,
+        providers,
+        path: {
+          index: -1,
+          queued: 0,
+          running: 0,
+          total: 0
+        }
+      }, options)
 
       found += providers.length
 
@@ -203,7 +243,7 @@ export class ContentRouting {
         key: target
       }
 
-      yield * self.network.sendRequest(peer, request, {
+      yield * self.network.sendRequest(peer.id, request, {
         ...options,
         signal,
         path
@@ -230,7 +270,11 @@ export class ContentRouting {
         }
 
         if (newProviders.length > 0) {
-          yield providerEvent({ from: event.from, providers: newProviders }, options)
+          yield providerEvent({
+            from: event.from,
+            providers: newProviders,
+            path: event.path
+          }, options)
 
           found += newProviders.length