fix: invoke onProgress callback with DHT queries during routing

package.json

@@ -143,12 +143,12 @@
     "it-drain": "^2.0.0",
     "it-filter": "^2.0.0",
     "it-first": "^2.0.0",
-    "it-parallel": "^3.0.0",
     "it-handshake": "^4.1.2",
     "it-length-prefixed": "^8.0.2",
     "it-map": "^2.0.0",
     "it-merge": "^2.0.0",
     "it-pair": "^2.0.2",
+    "it-parallel": "^3.0.0",
     "it-pb-stream": "^3.2.0",
     "it-pipe": "^2.0.3",
     "it-stream-types": "^1.0.4",
@@ -159,6 +159,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",
     "retimer": "^3.0.0",

src/dht/dht-content-routing.ts

@@ -1,39 +1,64 @@
-import drain from 'it-drain'
 import { CodeError } from '@libp2p/interfaces/errors'
-import type { DHT } from '@libp2p/interface-dht'
+import type { DHT, QueryEvent } from '@libp2p/interface-dht'
 import type { ContentRouting } from '@libp2p/interface-content-routing'
 import type { CID } from 'multiformats/cid'
 import type { AbortOptions } from '@libp2p/interfaces'
 import type { PeerInfo } from '@libp2p/interface-peer-info'
+import { CustomProgressEvent, type ProgressEvent, type 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>
 
 /**
  * Wrapper class to convert events into returned values
  */
-export class DHTContentRouting implements ContentRouting {
+export class DHTContentRouting implements ContentRouting<
+  ProvideProgressEvents,
+  FindProvidersProgressEvents,
+  PutProgressEvents,
+  GetProgressEvents
+> {
   private readonly dht: DHT
 
   constructor (dht: DHT) {
     this.dht = dht
   }
 
-  async provide (cid: CID): Promise<void> {
-    await drain(this.dht.provide(cid))
+  async provide (cid: CID, options: AbortOptions & 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: AbortOptions = {}): AsyncGenerator<PeerInfo, void, undefined> {
+  async * findProviders (cid: CID, options: AbortOptions & 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?: AbortOptions): Promise<void> {
-    await drain(this.dht.put(key, value, options))
+  async put (key: Uint8Array, value: Uint8Array, options: AbortOptions & 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?: AbortOptions): Promise<Uint8Array> {
+  async get (key: Uint8Array, options: AbortOptions & 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
       }

src/dht/dht-peer-routing.ts

@@ -1,23 +1,35 @@
 import { CodeError } from '@libp2p/interfaces/errors'
 import { messages, codes } from '../errors.js'
 import type { PeerRouting } from '@libp2p/interface-peer-routing'
-import type { DHT } from '@libp2p/interface-dht'
+import type { DHT, QueryEvent } from '@libp2p/interface-dht'
 import type { PeerId } from '@libp2p/interface-peer-id'
 import type { AbortOptions } from '@libp2p/interfaces'
 import type { PeerInfo } from '@libp2p/interface-peer-info'
+import { CustomProgressEvent, type ProgressEvent, type ProgressOptions } from 'progress-events'
+
+export type FindPeerProgressEvents =
+  ProgressEvent<'libp2p:peer-routing:find-peer:dht:event', QueryEvent>
+
+export type GetClosestPeerProgressEvents =
+  ProgressEvent<'libp2p:peer-routing:get-closest-peer:dht:event', QueryEvent>
 
 /**
  * Wrapper class to convert events into returned values
  */
-export class DHTPeerRouting implements PeerRouting {
+export class DHTPeerRouting implements PeerRouting<
+  FindPeerProgressEvents,
+  GetClosestPeerProgressEvents
+> {
   private readonly dht: DHT
 
   constructor (dht: DHT) {
     this.dht = dht
   }
 
-  async findPeer (peerId: PeerId, options: AbortOptions = {}): Promise<PeerInfo> {
+  async findPeer (peerId: PeerId, options: AbortOptions & 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
       }
@@ -26,8 +38,10 @@ export class DHTPeerRouting implements PeerRouting {
     throw new CodeError(messages.NOT_FOUND, codes.ERR_NOT_FOUND)
   }
 
-  async * getClosestPeers (key: Uint8Array, options: AbortOptions = {}): AsyncIterable<PeerInfo> {
+  async * getClosestPeers (key: Uint8Array, options: AbortOptions & ProgressOptions<GetClosestPeerProgressEvents> = {}): AsyncIterable<PeerInfo> {
     for await (const event of this.dht.getClosestPeers(key, options)) {
+      options.onProgress?.(new CustomProgressEvent('libp2p:peer-routing:get-closest-peer:dht:event', event))
+
       if (event.name === 'FINAL_PEER') {
         yield event.peer
       }

test/content-routing/content-routing.node.ts

@@ -115,6 +115,39 @@ describe('content-routing', () => {
 
       return await deferred.promise
     })
+
+    it('should call progress events', async () => {
+      const deferred = pDefer()
+
+      if (nodes[0].dht == null) {
+        throw new Error('DHT was not configured')
+      }
+
+      sinon.stub(nodes[0].dht, 'findProviders').callsFake(async function * () {
+        yield {
+          from: nodes[0].peerId,
+          type: 0,
+          name: 'PROVIDER',
+          providers: [{
+            id: nodes[0].peerId,
+            multiaddrs: [],
+            protocols: []
+          }]
+        }
+        deferred.resolve()
+      })
+
+      const onProgress = sinon.stub()
+
+      await drain(nodes[0].contentRouting.findProviders(CID.parse('QmU621oD8AhHw6t25vVyfYKmL9VV3PTgc52FngEhTGACFB'), {
+        // @ts-expect-error - can be removed when the content routing interface is updated to include progress events
+        onProgress
+      }))
+
+      await deferred.promise
+
+      expect(onProgress.called).to.be.true()
+    })
   })
 
   describe('via delegate router', () => {