feat: add path index to events (#3102)

KAD-DHT queries operate on disjoint paths which can make following
query events hard.

Add a `.path` property to events with the numeric index of the
disjoint path the query is following, this makes reconciling events
in progress handlers much easier.

Author: achingbrain

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

@@ -18,7 +18,7 @@ import { verifyRecord } from '../record/validators.js'
 import { createPutRecord, bufferToRecordKey } from '../utils.js'
 import type { KadDHTComponents, Validators, Selectors, ValueEvent, QueryEvent } from '../index.js'
 import type { Message } from '../message/dht.js'
-import type { Network } from '../network.js'
+import type { Network, SendMessageOptions } 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'
@@ -88,7 +88,7 @@ export class ContentFetching {
   /**
    * Send the best record found to any peers that have an out of date record
    */
-  async * sendCorrectionRecord (key: Uint8Array, vals: ValueEvent[], best: Uint8Array, options: RoutingOptions = {}): AsyncGenerator<QueryEvent> {
+  async * sendCorrectionRecord (key: Uint8Array, vals: ValueEvent[], best: Uint8Array, options: SendMessageOptions): AsyncGenerator<QueryEvent> {
     this.log('sendCorrection for %b', key)
     const fixupRec = createPutRecord(key, best)
 
@@ -139,7 +139,7 @@ export class ContentFetching {
   /**
    * Store the given key/value pair in the DHT
    */
-  async * put (key: Uint8Array, value: Uint8Array, options: RoutingOptions = {}): AsyncGenerator<unknown, void, undefined> {
+  async * put (key: Uint8Array, value: Uint8Array, options: RoutingOptions): AsyncGenerator<unknown, void, undefined> {
     this.log('put key %b value %b', key, value)
 
     // create record in the dht format
@@ -171,7 +171,10 @@ export class ContentFetching {
           }
 
           this.log('send put to %p', event.peer.id)
-          for await (const putEvent of this.network.sendRequest(event.peer.id, msg, options)) {
+          for await (const putEvent of this.network.sendRequest(event.peer.id, msg, {
+            ...options,
+            path: event.path
+          })) {
             events.push(putEvent)
 
             if (putEvent.name !== 'PEER_RESPONSE') {
@@ -201,7 +204,7 @@ export class ContentFetching {
   /**
    * Get the value to the given key
    */
-  async * get (key: Uint8Array, options: RoutingOptions = {}): AsyncGenerator<QueryEvent | ValueEvent> {
+  async * get (key: Uint8Array, options: RoutingOptions): AsyncGenerator<QueryEvent | ValueEvent> {
     this.log('get %b', key)
 
     const vals: ValueEvent[] = []
@@ -237,7 +240,10 @@ export class ContentFetching {
       throw new NotFoundError('Best value was not found')
     }
 
-    yield * this.sendCorrectionRecord(key, vals, best, options)
+    yield * this.sendCorrectionRecord(key, vals, best, {
+      ...options,
+      path: -1
+    })
 
     yield vals[i]
   }
@@ -261,10 +267,11 @@ export class ContentFetching {
 
     const self = this // eslint-disable-line @typescript-eslint/no-this-alias
 
-    const getValueQuery: QueryFunc = async function * ({ peer, signal }) {
+    const getValueQuery: QueryFunc = async function * ({ peer, signal, path }) {
       for await (const event of self.peerRouting.getValueOrPeers(peer, key, {
         ...options,
-        signal
+        signal,
+        path
       })) {
         yield event
 

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

@@ -112,7 +112,10 @@ export class ContentRouting {
         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)) {
+          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++
@@ -181,7 +184,7 @@ export class ContentRouting {
         }
       }
 
-      yield peerResponseEvent({ from: this.components.peerId, messageType: MessageType.GET_PROVIDERS, providers }, options)
+      yield peerResponseEvent({ from: this.components.peerId, messageType: MessageType.GET_PROVIDERS, providers, path: -1 }, options)
       yield providerEvent({ from: this.components.peerId, providers }, options)
 
       found += providers.length
@@ -194,15 +197,16 @@ export class ContentRouting {
     /**
      * The query function to use on this particular disjoint path
      */
-    const findProvidersQuery: QueryFunc = async function * ({ peer, signal }) {
+    const findProvidersQuery: QueryFunc = async function * ({ peer, signal, path }) {
       const request = {
         type: MessageType.GET_PROVIDERS,
         key: target
       }
 
       yield * self.network.sendRequest(peer, request, {
         ...options,
-        signal
+        signal,
+        path
       })
     }
 

packages/kad-dht/src/index.ts

@@ -192,6 +192,7 @@ export interface SendQueryEvent {
   name: 'SEND_QUERY'
   messageName: keyof typeof MessageType
   messageType: MessageType
+  path: number
 }
 
 /**
@@ -207,6 +208,7 @@ export interface PeerResponseEvent {
   closer: PeerInfo[]
   providers: PeerInfo[]
   record?: DHTRecord
+  path: number
 }
 
 /**
@@ -217,6 +219,7 @@ export interface FinalPeerEvent {
   peer: PeerInfo
   type: EventTypes.FINAL_PEER
   name: 'FINAL_PEER'
+  path: number
 }
 
 /**
@@ -227,6 +230,7 @@ export interface QueryErrorEvent {
   type: EventTypes.QUERY_ERROR
   name: 'QUERY_ERROR'
   error: Error
+  path?: number
 }
 
 /**
@@ -256,10 +260,13 @@ export interface AddPeerEvent {
   type: EventTypes.ADD_PEER
   name: 'ADD_PEER'
   peer: PeerId
+  path: number
 }
 
 /**
  * Emitted when peers are dialled as part of a query
+ *
+ * @deprecated No longer emitted as sometimes connections are reused so it's not possible to say with certainty that a peer has been dialled
  */
 export interface DialPeerEvent {
   peer: PeerId
@@ -506,8 +513,8 @@ export interface KadDHTInit {
   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.
+   * After startup by default all queries will be paused until there is at least
+   * one peer in the routing table.
    *
    * Pass true here to disable this behavior.
    *

packages/kad-dht/src/kad-dht.ts

@@ -223,7 +223,8 @@ export class KadDHT extends TypedEventEmitter<PeerDiscoveryEvents> implements Ka
       logPrefix,
       metricsPrefix,
       initialQuerySelfHasRun,
-      routingTable: this.routingTable
+      routingTable: this.routingTable,
+      allowQueryWithZeroPeers: init.allowQueryWithZeroPeers
     })
 
     // DHT components
@@ -276,7 +277,6 @@ export class KadDHT extends TypedEventEmitter<PeerDiscoveryEvents> implements Ka
       initialInterval: init.initialQuerySelfInterval,
       logPrefix,
       initialQuerySelfHasRun,
-      routingTable: this.routingTable,
       operationMetrics
     })
     this.reprovider = new Reprovider(components, {

packages/kad-dht/src/network.ts

@@ -5,7 +5,6 @@ import { pbStream } from 'it-protobuf-stream'
 import { Message } from './message/dht.js'
 import { fromPbPeerInfo } from './message/utils.js'
 import {
-  dialPeerEvent,
   sendQueryEvent,
   peerResponseEvent,
   queryErrorEvent
@@ -24,6 +23,15 @@ interface NetworkEvents {
   'peer': CustomEvent<PeerInfo>
 }
 
+export interface SendMessageOptions extends RoutingOptions {
+  /**
+   * Queries involve following up to `k` disjoint paths through the network -
+   * this option is which index within `k` this message is for, and it
+   * allows observers to collate events together on a per-path basis
+   */
+  path: number
+}
+
 /**
  * Handle network operations for the dht
  */
@@ -142,7 +150,7 @@ export class Network extends TypedEventEmitter<NetworkEvents> implements Startab
   /**
    * Send a request and read a response
    */
-  async * sendRequest (to: PeerId, msg: Partial<Message>, options: RoutingOptions = {}): AsyncGenerator<QueryEvent> {
+  async * sendRequest (to: PeerId, msg: Partial<Message>, options: SendMessageOptions): AsyncGenerator<QueryEvent> {
     if (!this.running) {
       return
     }
@@ -154,8 +162,7 @@ export class Network extends TypedEventEmitter<NetworkEvents> implements Startab
     }
 
     this.log('sending %s to %p', msg.type, to)
-    yield dialPeerEvent({ peer: to }, options)
-    yield sendQueryEvent({ to, type }, options)
+    yield sendQueryEvent({ to, type, path: options.path }, options)
 
     let stream: Stream | undefined
     const signal = this.timeout.getTimeoutSignal(options)
@@ -183,7 +190,8 @@ export class Network extends TypedEventEmitter<NetworkEvents> implements Startab
         messageType: response.type,
         closer: response.closer.map(fromPbPeerInfo),
         providers: response.providers.map(fromPbPeerInfo),
-        record: response.record == null ? undefined : Libp2pRecord.deserialize(response.record)
+        record: response.record == null ? undefined : Libp2pRecord.deserialize(response.record),
+        path: options.path
       }, options)
     } catch (err: any) {
       this.metrics.errors?.increment({ [type]: true })
@@ -196,7 +204,7 @@ export class Network extends TypedEventEmitter<NetworkEvents> implements Startab
         this.log.error('could not send %s to %p - %e', msg.type, to, err)
       }
 
-      yield queryErrorEvent({ from: to, error: err }, options)
+      yield queryErrorEvent({ from: to, error: err, path: options.path }, options)
     } finally {
       this.timeout.cleanUp(signal)
     }
@@ -205,7 +213,7 @@ export class Network extends TypedEventEmitter<NetworkEvents> implements Startab
   /**
    * Sends a message without expecting an answer
    */
-  async * sendMessage (to: PeerId, msg: Partial<Message>, options: RoutingOptions = {}): AsyncGenerator<QueryEvent> {
+  async * sendMessage (to: PeerId, msg: Partial<Message>, options: SendMessageOptions): AsyncGenerator<QueryEvent> {
     if (!this.running) {
       return
     }
@@ -217,8 +225,7 @@ export class Network extends TypedEventEmitter<NetworkEvents> implements Startab
     }
 
     this.log('sending %s to %p', msg.type, to)
-    yield dialPeerEvent({ peer: to }, options)
-    yield sendQueryEvent({ to, type }, options)
+    yield sendQueryEvent({ to, type, path: options.path }, options)
 
     let stream: Stream | undefined
     const signal = this.timeout.getTimeoutSignal(options)
@@ -242,12 +249,12 @@ export class Network extends TypedEventEmitter<NetworkEvents> implements Startab
           stream?.abort(err)
         })
 
-      yield peerResponseEvent({ from: to, messageType: type }, options)
+      yield peerResponseEvent({ from: to, messageType: type, path: options.path }, options)
     } catch (err: any) {
       this.metrics.errors?.increment({ [type]: true })
 
       stream?.abort(err)
-      yield queryErrorEvent({ from: to, error: err }, options)
+      yield queryErrorEvent({ from: to, error: err, path: options.path }, options)
     } finally {
       this.timeout.cleanUp(signal)
     }

packages/kad-dht/src/peer-distance-list.ts

@@ -6,6 +6,7 @@ import type { PeerId, PeerInfo } from '@libp2p/interface'
 interface PeerDistance {
   peer: PeerInfo
   distance: Uint8Array
+  path: number
 }
 
 /**
@@ -40,30 +41,31 @@ export class PeerDistanceList {
   /**
    * The peers in the list, in order of distance from the origin key
    */
-  get peers (): PeerInfo[] {
-    return this.peerDistances.map(pd => pd.peer)
+  get peers (): PeerDistance[] {
+    return [...this.peerDistances]
   }
 
   /**
    * Add a peerId to the list.
    */
-  async add (peer: PeerInfo): Promise<void> {
+  async add (peer: PeerInfo, path: number = -1): Promise<void> {
     const dhtKey = await convertPeerId(peer.id)
 
-    this.addWithKadId(peer, dhtKey)
+    this.addWithKadId(peer, dhtKey, path)
   }
 
   /**
    * Add a peerId to the list.
    */
-  addWithKadId (peer: PeerInfo, kadId: Uint8Array): void {
+  addWithKadId (peer: PeerInfo, kadId: Uint8Array, path: number = -1): void {
     if (this.peerDistances.find(pd => pd.peer.id.equals(peer.id)) != null) {
       return
     }
 
-    const el = {
+    const el: PeerDistance = {
       peer,
-      distance: uint8ArrayXor(this.originDhtKey, kadId)
+      distance: uint8ArrayXor(this.originDhtKey, kadId),
+      path
     }
 
     let added = false