fix: count dht providers and abort when enough found (#2470)

The way we were counting DHT providers was wrong, count the actual
number found - this prevents queries continuing to execute after
we've found K providers.

Author: achingbrain

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

@@ -128,6 +128,7 @@ export class ContentRouting {
    */
   async * findProviders (key: CID, options: RoutingOptions): AsyncGenerator<PeerResponseEvent | ProviderEvent | QueryEvent> {
     const toFind = this.routingTable.kBucketSize
+    let found = 0
     const target = key.multihash.bytes
     const self = this // eslint-disable-line @typescript-eslint/no-this-alias
 
@@ -158,11 +159,12 @@ export class ContentRouting {
 
       yield peerResponseEvent({ from: this.components.peerId, messageType: MessageType.GET_PROVIDERS, providers }, options)
       yield providerEvent({ from: this.components.peerId, providers }, options)
-    }
 
-    // All done
-    if (provs.length >= toFind) {
-      return
+      found += providers.length
+
+      if (found >= toFind) {
+        return
+      }
     }
 
     /**
@@ -201,10 +203,12 @@ export class ContentRouting {
 
         if (newProviders.length > 0) {
           yield providerEvent({ from: event.from, providers: newProviders }, options)
-        }
 
-        if (providers.size === toFind) {
-          return
+          found += newProviders.length
+
+          if (found >= toFind) {
+            return
+          }
         }
       }
     }

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

@@ -106,10 +106,11 @@ export class QuerySelf implements Startable {
 
     if (this.started) {
       this.controller = new AbortController()
-      const signal = anySignal([this.controller.signal, AbortSignal.timeout(this.queryTimeout)])
+      const timeoutSignal = AbortSignal.timeout(this.queryTimeout)
+      const signal = anySignal([this.controller.signal, timeoutSignal])
 
       // this controller will get used for lots of dial attempts so make sure we don't cause warnings to be logged
-      setMaxListeners(Infinity, signal)
+      setMaxListeners(Infinity, signal, this.controller.signal, timeoutSignal)
 
       try {
         if (this.routingTable.size === 0) {

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

@@ -48,7 +48,7 @@ export interface QueryOptions extends RoutingOptions {
 export class QueryManager implements Startable {
   public disjointPaths: number
   private readonly alpha: number
-  private readonly shutDownController: AbortController
+  private shutDownController: AbortController
   private running: boolean
   private queries: number
   private readonly logger: ComponentLogger
@@ -96,6 +96,11 @@ export class QueryManager implements Startable {
    */
   async start (): Promise<void> {
     this.running = true
+
+    // allow us to stop queries on shut down
+    this.shutDownController = new AbortController()
+    // make sure we don't make a lot of noise in the logs
+    setMaxListeners(Infinity, this.shutDownController.signal)
   }
 
   /**
@@ -131,7 +136,6 @@ export class QueryManager implements Startable {
     // if the user breaks out of a for..await of loop iterating over query
     // results we need to cancel any in-flight network requests
     const queryEarlyExitController = new AbortController()
-    setMaxListeners(Infinity, queryEarlyExitController.signal)
 
     const signal = anySignal([
       this.shutDownController.signal,
@@ -141,7 +145,7 @@ export class QueryManager implements Startable {
 
     // this signal will get listened to for every invocation of queryFunc
     // so make sure we don't make a lot of noise in the logs
-    setMaxListeners(Infinity, signal)
+    setMaxListeners(Infinity, signal, queryEarlyExitController.signal)
 
     const log = this.logger.forComponent(`${this.logPrefix}:query:` + uint8ArrayToString(key, 'base58btc'))
 

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

@@ -1,3 +1,4 @@
+import { setMaxListeners } from '@libp2p/interface'
 import { anySignal } from 'any-signal'
 import Queue from 'p-queue'
 import { toString } from 'uint8arrays/to-string'
@@ -112,6 +113,9 @@ export async function * queryPath (options: QueryPathOptions): AsyncGenerator<Qu
 
       const compoundSignal = anySignal(signals)
 
+      // this signal can get listened to a lot
+      setMaxListeners(Infinity, compoundSignal)
+
       try {
         for await (const event of query({
           key,