Merge branch 'master' into biome

Author: acolytec3

packages/portalnetwork/src/networks/contentLookup.ts

@@ -25,6 +25,7 @@ export class ContentLookup {
   private finished: boolean
   private content: ContentLookupResponse
   private pending: Set<NodeId>
+  private queuedPeers: Set<NodeId>
   private completedRequests?: Map<NodeId, NodeId[]>
   private contentTrace?: ContentTrace
   constructor(network: BaseNetwork, contentKey: Uint8Array, trace = false) {
@@ -37,6 +38,7 @@ export class ContentLookup {
     this.finished = false
     this.meta = new Map()
     this.pending = new Set()
+    this.queuedPeers = new Set()
     this.completedRequests = trace ? new Map() : undefined
     this.contentTrace = trace
       ? {
@@ -47,6 +49,23 @@ export class ContentLookup {
       : undefined
   }
 
+  private addPeerToQueue = (enr: ENR) => {
+    if (this.queuedPeers.has(enr.nodeId) || this.network.portal.uTP.hasRequests(enr.nodeId)) {
+      return
+    }
+    
+    const dist = distance(enr.nodeId, this.contentId)
+    this.lookupPeers.push({ enr, distance: Number(dist) })
+    this.queuedPeers.add(enr.nodeId)
+    this.meta.set('0x' + enr.nodeId, {
+      enr: enr.encodeTxt(),
+      distance: bigIntToHex(dist),
+    })
+    this.logger(
+      `Adding ${shortId(enr.nodeId)} to lookup queue (${this.lookupPeers.size()})`,
+    )
+  }
+
   /**
    * Queries the 5 nearest nodes in the history network routing table and recursively
    * requests peers closer to the content until either the content is found or there are no more peers to query
@@ -73,26 +92,28 @@ export class ContentLookup {
     // Sort known peers by distance to the content
     const nearest = this.network.routingTable.values()
     for (const enr of nearest) {
-      // // Skip if the node has an active uTP request
-      if (this.network.portal.uTP.hasRequests(enr.nodeId) === true) {
-        continue
-      }
-      const dist = distance(enr.nodeId, this.contentId)
-      this.lookupPeers.push({ enr, distance: Number(dist) })
-      this.meta.set(enr.nodeId, { enr: enr.encodeTxt(), distance: bigIntToHex(dist) })
+      this.addPeerToQueue(enr)
     }
 
     while (!this.finished && (this.lookupPeers.length > 0 || this.pending.size > 0)) {
       if (this.lookupPeers.length > 0) {
         // Ask more peers (up to 5) for content
         const peerBatch: LookupPeer[] = []
-        while (this.lookupPeers.peek() && peerBatch.length < 5) {
+        const availableSlots = 5 - this.pending.size
+        while (this.lookupPeers.peek() && peerBatch.length < availableSlots) {
           const next = this.lookupPeers.pop()!
           peerBatch.push(next)
         }
         const promises = peerBatch.map((peer) => {
+          const controller = new AbortController()
+          const timeoutId = setTimeout(() => {
+            controller.abort()
+          }, this.timeout)
+          
           return Promise.race([
-            this.processPeer(peer),
+            this.processPeer(peer, controller.signal).finally(() => {
+              clearTimeout(timeoutId)
+            }),
             new Promise((resolve) =>
               setTimeout(() => {
                 resolve(undefined)
@@ -155,7 +176,7 @@ export class ContentLookup {
     return this.content
   }
 
-  private processPeer = async (peer: LookupPeer): Promise<ContentLookupResponse | void> => {
+  private processPeer = async (peer: LookupPeer, signal?: AbortSignal): Promise<ContentLookupResponse | void> => {
     if (this.finished) return
     if (this.network.routingTable.isIgnored(peer.enr.nodeId)) {
       this.logger(`peer ${shortId(peer.enr.nodeId)} is ignored`)
@@ -165,7 +186,21 @@ export class ContentLookup {
     this.pending.add(peer.enr.encodeTxt())
     this.logger(`Requesting content from ${shortId(peer.enr.nodeId)}`)
     try {
-      const res = await this.network.sendFindContent!(peer.enr, this.contentKey)
+      // Create a promise that rejects when the signal is aborted
+      const abortPromise = new Promise((_, reject) => {
+        if (signal) {
+          signal.addEventListener('abort', () => {
+            reject(new Error('Request cancelled'))
+          })
+        }
+      })
+
+      // Race between the actual request and the abort signal
+      const res = await Promise.race([
+        this.network.sendFindContent!(peer.enr, this.contentKey),
+        abortPromise
+      ]) as ContentLookupResponse | undefined
+
       this.pending.delete(peer.enr.encodeTxt())
       if (this.finished) {
         this.logger(`Response from ${shortId(peer.enr.nodeId)} arrived after lookup finished`)
@@ -193,21 +228,7 @@ export class ContentLookup {
         this.logger(`received ${res.enrs.length} ENRs for closer nodes`)
         for (const enr of res.enrs) {
           const decodedEnr = ENR.decode(enr)
-          // // Skip if the node has an active uTP request
-          if (this.network.portal.uTP.hasRequests(decodedEnr.nodeId) === true) {
-            continue
-          }
-          if (!this.meta.has(decodedEnr.nodeId)) {
-            const dist = distance(decodedEnr.nodeId, this.contentId)
-            this.lookupPeers.push({ enr: decodedEnr, distance: Number(dist) })
-            this.meta.set('0x' + decodedEnr.nodeId, {
-              enr: decodedEnr.encodeTxt(),
-              distance: bigIntToHex(dist),
-            })
-            this.logger(
-              `Adding ${shortId(decodedEnr.nodeId)} to lookup queue (${this.lookupPeers.size()})`,
-            )
-          }
+          this.addPeerToQueue(decodedEnr)
         }
         this.completedRequests &&
           this.completedRequests.set(
@@ -218,6 +239,9 @@ export class ContentLookup {
       }
     } catch (err) {
       this.pending.delete(peer.enr.encodeTxt())
+      if (signal?.aborted) {
+        this.logger(`Request to ${shortId(peer.enr.nodeId)} was cancelled`)
+      }
       throw err
     }
   }