fix: invoke onProgress callback with DHT queries during routing

Allow passing an `onProgress` callback to the peer/content routers that can receive DHT query events.

Refs #1574

Author: achingbrain

packages/interface/package.json

@@ -163,6 +163,7 @@
     "it-stream-types": "^2.0.1",
     "multiformats": "^12.0.1",
     "p-defer": "^4.0.0",
+    "progress-events": "^1.0.0",
     "uint8arraylist": "^2.4.3"
   },
   "devDependencies": {

packages/interface/src/content-routing/index.ts

@@ -1,6 +1,7 @@
 import type { AbortOptions } from '../index.js'
 import type { PeerInfo } from '../peer-info/index.js'
 import type { CID } from 'multiformats/cid'
+import type { ProgressEvent, ProgressOptions } from 'progress-events'
 
 /**
  * Any object that implements this Symbol as a property should return a
@@ -23,7 +24,12 @@ import type { CID } from 'multiformats/cid'
  */
 export const contentRouting = Symbol.for('@libp2p/content-routing')
 
-export interface ContentRouting {
+export interface ContentRouting<
+  ProvideProgressEvents extends ProgressEvent = ProgressEvent,
+  FindProvidersProgressEvents extends ProgressEvent = ProgressEvent,
+  PutProgressEvents extends ProgressEvent = ProgressEvent,
+  GetProgressEvents extends ProgressEvent = ProgressEvent
+> {
   /**
    * The implementation of this method should ensure that network peers know the
    * caller can provide content that corresponds to the passed CID.
@@ -35,7 +41,7 @@ export interface ContentRouting {
    * await contentRouting.provide(cid)
    * ```
    */
-  provide: (cid: CID, options?: AbortOptions) => Promise<void>
+  provide: (cid: CID, options?: AbortOptions & ProgressOptions<ProvideProgressEvents>) => Promise<void>
 
   /**
    * Find the providers of the passed CID.
@@ -49,7 +55,7 @@ export interface ContentRouting {
    * }
    * ```
    */
-  findProviders: (cid: CID, options?: AbortOptions) => AsyncIterable<PeerInfo>
+  findProviders: (cid: CID, options?: AbortOptions & ProgressOptions<FindProvidersProgressEvents>) => AsyncIterable<PeerInfo>
 
   /**
    * Puts a value corresponding to the passed key in a way that can later be
@@ -65,7 +71,7 @@ export interface ContentRouting {
    * await contentRouting.put(key, value)
    * ```
    */
-  put: (key: Uint8Array, value: Uint8Array, options?: AbortOptions) => Promise<void>
+  put: (key: Uint8Array, value: Uint8Array, options?: AbortOptions & ProgressOptions<PutProgressEvents>) => Promise<void>
 
   /**
    * Retrieves a value from the network corresponding to the passed key.
@@ -79,5 +85,5 @@ export interface ContentRouting {
    * const value = await contentRouting.get(key)
    * ```
    */
-  get: (key: Uint8Array, options?: AbortOptions) => Promise<Uint8Array>
+  get: (key: Uint8Array, options?: AbortOptions & ProgressOptions<GetProgressEvents>) => Promise<Uint8Array>
 }

packages/interface/src/index.ts

@@ -28,6 +28,7 @@ import type { StreamHandler, StreamHandlerOptions } from './stream-handler/index
 import type { Topology } from './topology/index.js'
 import type { Listener } from './transport/index.js'
 import type { Multiaddr } from '@multiformats/multiaddr'
+import type { ProgressEvent } from 'progress-events'
 
 /**
  * Used by the connection manager to sort addresses into order before dialling
@@ -108,7 +109,15 @@ export interface IdentifyResult {
  * Event names are `noun:verb` so the first part is the name of the object
  * being acted on and the second is the action.
  */
-export interface Libp2pEvents<T extends ServiceMap = ServiceMap> {
+export interface Libp2pEvents<
+  Services extends ServiceMap = ServiceMap,
+  FindPeerProgressEvents extends ProgressEvent = ProgressEvent,
+  GetClosestPeersProgressEvents extends ProgressEvent = ProgressEvent,
+  ProvideProgressEvents extends ProgressEvent = ProgressEvent,
+  FindProvidersProgressEvents extends ProgressEvent = ProgressEvent,
+  PutProgressEvents extends ProgressEvent = ProgressEvent,
+  GetProgressEvents extends ProgressEvent = ProgressEvent
+> {
   /**
    * This event is dispatched when a new network peer is discovered.
    *
@@ -235,7 +244,15 @@ export interface Libp2pEvents<T extends ServiceMap = ServiceMap> {
    * })
    * ```
    */
-  'start': CustomEvent<Libp2p<T>>
+  'start': CustomEvent<Libp2p<
+  Services,
+  FindPeerProgressEvents,
+  GetClosestPeersProgressEvents,
+  ProvideProgressEvents,
+  FindProvidersProgressEvents,
+  PutProgressEvents,
+  GetProgressEvents
+  >>
 
   /**
    * This event notifies listeners that the node has stopped
@@ -246,7 +263,15 @@ export interface Libp2pEvents<T extends ServiceMap = ServiceMap> {
    * })
    * ```
    */
-  'stop': CustomEvent<Libp2p<T>>
+  'stop': CustomEvent<Libp2p<
+  Services,
+  FindPeerProgressEvents,
+  GetClosestPeersProgressEvents,
+  ProvideProgressEvents,
+  FindProvidersProgressEvents,
+  PutProgressEvents,
+  GetProgressEvents
+  >>
 }
 
 /**
@@ -303,7 +328,23 @@ export interface PendingDial {
 /**
  * Libp2p nodes implement this interface.
  */
-export interface Libp2p<T extends ServiceMap = ServiceMap> extends Startable, EventEmitter<Libp2pEvents<T>> {
+export interface Libp2p<
+  Services extends ServiceMap = ServiceMap,
+  FindPeerProgressEvents extends ProgressEvent = ProgressEvent,
+  GetClosestPeersProgressEvents extends ProgressEvent = ProgressEvent,
+  ProvideProgressEvents extends ProgressEvent = ProgressEvent,
+  FindProvidersProgressEvents extends ProgressEvent = ProgressEvent,
+  PutProgressEvents extends ProgressEvent = ProgressEvent,
+  GetProgressEvents extends ProgressEvent = ProgressEvent
+> extends Startable, EventEmitter<Libp2pEvents<
+  Services,
+  FindPeerProgressEvents,
+  GetClosestPeersProgressEvents,
+  ProvideProgressEvents,
+  FindProvidersProgressEvents,
+  PutProgressEvents,
+  GetProgressEvents
+  >> {
   /**
    * The PeerId is a unique identifier for a node on the network.
    *
@@ -354,7 +395,7 @@ export interface Libp2p<T extends ServiceMap = ServiceMap> extends Startable, Ev
    * }
    * ```
    */
-  peerRouting: PeerRouting
+  peerRouting: PeerRouting<FindPeerProgressEvents, GetClosestPeersProgressEvents>
 
   /**
    * The content routing subsystem allows the user to find providers for content,
@@ -370,7 +411,7 @@ export interface Libp2p<T extends ServiceMap = ServiceMap> extends Startable, Ev
    * }
    * ```
    */
-  contentRouting: ContentRouting
+  contentRouting: ContentRouting<ProvideProgressEvents, FindProvidersProgressEvents, GetProgressEvents, PutProgressEvents>
 
   /**
    * The keychain contains the keys used by the current node, and can create new
@@ -597,7 +638,7 @@ export interface Libp2p<T extends ServiceMap = ServiceMap> extends Startable, Ev
   /**
    * A set of user defined services
    */
-  services: T
+  services: Services
 }
 
 /**

packages/interface/src/peer-routing/index.ts

@@ -1,6 +1,7 @@
 import type { AbortOptions } from '../index.js'
 import type { PeerId } from '../peer-id/index.js'
 import type { PeerInfo } from '../peer-info/index.js'
+import type { ProgressEvent, ProgressOptions } from 'progress-events'
 
 /**
  * Any object that implements this Symbol as a property should return a
@@ -23,7 +24,10 @@ import type { PeerInfo } from '../peer-info/index.js'
  */
 export const peerRouting = Symbol.for('@libp2p/peer-routing')
 
-export interface PeerRouting {
+export interface PeerRouting<
+  FindPeerProgressEvents extends ProgressEvent = ProgressEvent,
+  GetClosestPeersProgressEvents extends ProgressEvent = ProgressEvent
+> {
   /**
    * Searches the network for peer info corresponding to the passed peer id.
    *
@@ -34,7 +38,7 @@ export interface PeerRouting {
    * const peer = await peerRouting.findPeer(peerId, options)
    * ```
    */
-  findPeer: (peerId: PeerId, options?: AbortOptions) => Promise<PeerInfo>
+  findPeer: (peerId: PeerId, options?: AbortOptions & ProgressOptions<FindPeerProgressEvents>) => Promise<PeerInfo>
 
   /**
    * Search the network for peers that are closer to the passed key. Peer
@@ -49,5 +53,5 @@ export interface PeerRouting {
    * }
    * ```
    */
-  getClosestPeers: (key: Uint8Array, options?: AbortOptions) => AsyncIterable<PeerInfo>
+  getClosestPeers: (key: Uint8Array, options?: AbortOptions & ProgressOptions<GetClosestPeersProgressEvents>) => AsyncIterable<PeerInfo>
 }

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

@@ -4,47 +4,73 @@ import { EventEmitter, CustomEvent } from '@libp2p/interface/events'
 import { type PeerDiscovery, peerDiscovery, type PeerDiscoveryEvents } from '@libp2p/interface/peer-discovery'
 import { type PeerRouting, peerRouting } from '@libp2p/interface/peer-routing'
 import { logger } from '@libp2p/logger'
-import drain from 'it-drain'
 import merge from 'it-merge'
 import isPrivate from 'private-ip'
+import { CustomProgressEvent } from 'progress-events'
 import { DefaultKadDHT } from './kad-dht.js'
 import { queryErrorEvent } from './query/events.js'
 import type { DualKadDHT, KadDHT, KadDHTComponents, KadDHTInit, QueryEvent, QueryOptions } from './index.js'
 import type { PeerId } from '@libp2p/interface/peer-id'
 import type { PeerInfo } from '@libp2p/interface/peer-info'
 import type { Multiaddr } from '@multiformats/multiaddr'
 import type { CID } from 'multiformats/cid'
+import type { ProgressEvent, ProgressOptions } from 'progress-events'
+
+export type ProvideProgressEvents =
+  ProgressEvent<'libp2p:content-routing:provide:dht:event', QueryEvent>
+
+export type FindProvidersProgressEvents =
+  ProgressEvent<'libp2p:content-routing:find-providers:dht:event', QueryEvent>
+
+export type PutProgressEvents =
+  ProgressEvent<'libp2p:content-routing:put:dht:event', QueryEvent>
+
+export type GetProgressEvents =
+  ProgressEvent<'libp2p:content-routing:get:dht:event', QueryEvent>
 
 const log = logger('libp2p:kad-dht')
 
 /**
  * Wrapper class to convert events into returned values
  */
-class DHTContentRouting implements ContentRouting {
+class DHTContentRouting implements ContentRouting<
+ProvideProgressEvents,
+FindProvidersProgressEvents,
+PutProgressEvents,
+GetProgressEvents
+> {
   private readonly dht: KadDHT
 
   constructor (dht: KadDHT) {
     this.dht = dht
   }
 
-  async provide (cid: CID, options: QueryOptions = {}): Promise<void> {
-    await drain(this.dht.provide(cid, options))
+  async provide (cid: CID, options: QueryOptions & ProgressOptions<ProvideProgressEvents> = {}): Promise<void> {
+    for await (const event of this.dht.provide(cid, options)) {
+      options.onProgress?.(new CustomProgressEvent('libp2p:content-routing:provide:dht:event', event))
+    }
   }
 
-  async * findProviders (cid: CID, options: QueryOptions = {}): AsyncGenerator<PeerInfo, void, undefined> {
+  async * findProviders (cid: CID, options: QueryOptions & ProgressOptions<FindProvidersProgressEvents> = {}): AsyncGenerator<PeerInfo, void, undefined> {
     for await (const event of this.dht.findProviders(cid, options)) {
+      options.onProgress?.(new CustomProgressEvent('libp2p:content-routing:find-providers:dht:event', event))
+
       if (event.name === 'PROVIDER') {
         yield * event.providers
       }
     }
   }
 
-  async put (key: Uint8Array, value: Uint8Array, options?: QueryOptions): Promise<void> {
-    await drain(this.dht.put(key, value, options))
+  async put (key: Uint8Array, value: Uint8Array, options: QueryOptions & ProgressOptions<PutProgressEvents> = {}): Promise<void> {
+    for await (const event of this.dht.put(key, value, options)) {
+      options.onProgress?.(new CustomProgressEvent('libp2p:content-routing:put:dht:event', event))
+    }
   }
 
-  async get (key: Uint8Array, options?: QueryOptions): Promise<Uint8Array> {
+  async get (key: Uint8Array, options: QueryOptions & ProgressOptions<GetProgressEvents> = {}): Promise<Uint8Array> {
     for await (const event of this.dht.get(key, options)) {
+      options.onProgress?.(new CustomProgressEvent('libp2p:content-routing:get:dht:event', event))
+
       if (event.name === 'VALUE') {
         return event.value
       }
@@ -54,18 +80,29 @@ class DHTContentRouting implements ContentRouting {
   }
 }
 
+export type FindPeerProgressEvents =
+  ProgressEvent<'libp2p:peer-routing:find-peer:dht:event', QueryEvent>
+
+export type GetClosestPeersProgressEvents =
+  ProgressEvent<'libp2p:peer-routing:get-closest-peers:dht:event', QueryEvent>
+
 /**
  * Wrapper class to convert events into returned values
  */
-class DHTPeerRouting implements PeerRouting {
+class DHTPeerRouting implements PeerRouting<
+FindPeerProgressEvents,
+GetClosestPeersProgressEvents
+> {
   private readonly dht: KadDHT
 
   constructor (dht: KadDHT) {
     this.dht = dht
   }
 
-  async findPeer (peerId: PeerId, options: QueryOptions = {}): Promise<PeerInfo> {
+  async findPeer (peerId: PeerId, options: QueryOptions & ProgressOptions<FindPeerProgressEvents> = {}): Promise<PeerInfo> {
     for await (const event of this.dht.findPeer(peerId, options)) {
+      options.onProgress?.(new CustomProgressEvent('libp2p:peer-routing:find-peer:dht:event', event))
+
       if (event.name === 'FINAL_PEER') {
         return event.peer
       }
@@ -74,8 +111,10 @@ class DHTPeerRouting implements PeerRouting {
     throw new CodeError('Not found', 'ERR_NOT_FOUND')
   }
 
-  async * getClosestPeers (key: Uint8Array, options: QueryOptions = {}): AsyncIterable<PeerInfo> {
+  async * getClosestPeers (key: Uint8Array, options: QueryOptions & ProgressOptions<GetClosestPeersProgressEvents> = {}): AsyncIterable<PeerInfo> {
     for await (const event of this.dht.getClosestPeers(key, options)) {
+      options.onProgress?.(new CustomProgressEvent('libp2p:peer-routing:get-closest-peers:dht:event', event))
+
       if (event.name === 'FINAL_PEER') {
         yield event.peer
       }

packages/libp2p/package.json

@@ -158,6 +158,7 @@
     "p-queue": "^7.3.4",
     "p-retry": "^5.0.0",
     "private-ip": "^3.0.0",
+    "progress-events": "^1.0.0",
     "protons-runtime": "^5.0.0",
     "rate-limiter-flexible": "^2.3.11",
     "uint8arraylist": "^2.4.3",