fix: limit concurrent HTTP requests

All HTTP requests made by this module are sent to the same delegate
host. Browsers throttle the number of concurrent requests per hostname,
right now it is 6 per host, which suffocates the use of delegate and
blocking it from being used for preload or delegated peer routing.

This change introduces task queues that limit the number of concurrent
requests, making it safe to run in browser context.

Optimizations:
- preload calls via `refs` are known to take time,
  so we limit them to max two at a time
- swarm.connect was the main offender (causing multiple requests to delegate),
  we now check if connection is already present and cache result for 1 minute
  removing most of redundant http requests
- hostname of default delegate is changed

Context: libp2p#12
Closes libp2p#12

License: MIT
Signed-off-by: Marcin Rataj <lidel@lidel.org>

.gitignore

@@ -61,4 +61,5 @@ typings/
 .env
 
 yarn.lock
-package-lock.json
+package-lock.json
+dist/

package.json

@@ -27,8 +27,11 @@
     "peer-id": "~0.13.1"
   },
   "dependencies": {
+    "debug": "^4.1.1",
     "ipfs-http-client": "^33.1.0",
-    "multiaddr": "^6.1.0"
+    "multiaddr": "^6.1.0",
+    "p-memoize": "^3.1.0",
+    "p-queue": "^6.1.0"
   },
   "contributors": [
     "Alan Shaw <alan.shaw@protocol.ai>",

src/index.js

@@ -5,14 +5,21 @@ const swarm = require('ipfs-http-client/src/swarm')
 const refs = require('ipfs-http-client/src/files-regular/refs')
 const defaultConfig = require('ipfs-http-client/src/utils/default-config')
 const multiaddr = require('multiaddr')
+const { default: PQueue } = require('p-queue')
+const pMemoize = require('p-memoize')
+const debug = require('debug')
+
+const log = debug('libp2p-delegated-content-routing')
+log.error = debug('libp2p-delegated-content-routing:error')
 
 const DEFAULT_MAX_TIMEOUT = 30e3 // 30 second default
 const DEFAULT_IPFS_API = {
   protocol: 'https',
   port: 443,
-  host: 'ipfs.io'
+  host: 'node0.delegate.ipfs.io'
 }
 
+// assuming below nodes need have autorelay enabled
 const DEFAULT_BOOSTRAP_NODES = [
   '/ipfs/QmSoLer265NRgSp2LA3dPaeykiS1J6DifTC88f5uVQKNAd',
   '/ipfs/QmSoLMeWqB7YGVLJN3pNLQpmmEk35v6wYtsMGLzSr5QBU3',
@@ -24,6 +31,9 @@ const DEFAULT_BOOSTRAP_NODES = [
   '/ipfs/Qmbut9Ywz9YEDrz8ySBSgWyJk41Uvm2QJPhwDJzJyGFsD6'
 ]
 
+const CONCURRENT_HTTP_REQUESTS = 4
+const SWARM_CONNECT_MAX_AGE = 60e3
+
 /**
  * An implementation of content routing, using a delegated peer.
  */
@@ -43,10 +53,26 @@ class DelegatedContentRouting {
     this.api = Object.assign({}, defaultConfig(), DEFAULT_IPFS_API, api)
     this.dht = dht(this.api)
     this.swarm = swarm(this.api)
+    // optimization: avoid calling swarm.connect too often
+    this.swarm.connect = pMemoize(this.swarm.connect, { maxAge: SWARM_CONNECT_MAX_AGE })
     this.refs = refs(this.api)
-
     this.peerId = peerId
-    this.bootstrappers = bootstrappers || DEFAULT_BOOSTRAP_NODES.map((addr) => multiaddr(addr))
+
+    bootstrappers = bootstrappers || DEFAULT_BOOSTRAP_NODES.map((addr) => multiaddr(addr))
+    this.circuits = bootstrappers.map((addr) => {
+      return addr.encapsulate(`/p2p-circuit/ipfs/${this.peerId.toB58String()}`)
+    })
+
+    // limit concurrency to avoid request flood in web browser
+    // https://github.com/libp2p/js-libp2p-delegated-content-routing/issues/12
+    const concurrency = { concurrency: CONCURRENT_HTTP_REQUESTS }
+    this._httpQueue = new PQueue(concurrency)
+    // sometimes refs requests take long time, they need separate queue
+    // to not suffocate regular bussiness
+    this._httpQueueRefs = new PQueue(Object.assign({}, concurrency, {
+      concurrency: 2
+    }))
+    log(`enabled DelegatedContentRouting via ${this.api.protocol}://${this.api.host}:${this.api.port}`)
   }
 
   /**
@@ -60,49 +86,66 @@ class DelegatedContentRouting {
    * @returns {AsyncIterable<PeerInfo>}
    */
   async * findProviders (key, options = {}) {
+    const keyString = key.toBaseEncodedString()
+    log('findProviders starts: ' + keyString)
     options.maxTimeout = options.maxTimeout || DEFAULT_MAX_TIMEOUT
 
-    const results = await this.dht.findProvs(key, {
+    const results = await this._httpQueue.add(() => this.dht.findProvs(key, {
       timeout: `${options.maxTimeout}ms` // The api requires specification of the time unit (s/ms)
-    })
+    }))
 
     for (let i = 0; i < results.length; i++) {
       yield results[i]
     }
+    log('findProviders finished: ' + keyString)
   }
 
   /**
    * Announce to the network that the delegated node can provide the given key.
    *
    * Currently this uses the following hack
    * - call swarm.connect on the delegated node to us, to ensure we are connected
-   * - call refs --recursive on the delegated node, so it fetches the content
+   * - call refs on the delegated node, so it fetches the content
    *
    * @param {CID} key
    * @param {function(Error)} callback
    * @returns {Promise<void>}
    */
   async provide (key) {
-    const addrs = this.bootstrappers.map((addr) => {
-      return addr.encapsulate(`/p2p-circuit/ipfs/${this.peerId.toB58String()}`)
-    })
-
-    const results = await Promise.all(
-      addrs.map((addr) => {
-        return this.swarm.connect(addr.toString()).catch(() => {})
-      })
-    )
+    const keyString = key.toBaseEncodedString()
+    log('provide starts: ' + keyString)
 
+    let results
+    try {
+      // optimization: try the first addr
+      // (swarm.connect will return success if ANY connection to this.peerId already exists)
+      const addr = this.circuits.find(a => !!a)
+      const res = await this._httpQueue.add(() => this.swarm.connect(addr.toString()))
+      if (res && res.error) throw new Error() // trigger fallback
+      results = [res]
+    } catch (err) {
+      // fallback to trying all potential circuits
+      results = await Promise.all(
+        this.circuits.map((addr) =>
+          this._httpQueue.add(() =>
+            this.swarm.connect(addr.toString()).catch(() => {})
+          )
+        )
+      )
+    }
     // only some need to succeed
     const success = results.filter((res) => res && res.error == null)
 
     if (success.length === 0) {
       throw new Error('unable to swarm.connect using p2p-circuit')
     }
 
-    this.refs(key.toBaseEncodedString(), {
-      recursive: true
-    })
+    // async preload of data to delegate node
+    // note: we call `provide` for every block, so it does not need to be recursive
+    await this._httpQueueRefs.add(() =>
+      this.refs(keyString, { recursive: false })
+    )
+    log('provide finished: ' + keyString)
   }
 }
 

test/index.spec.js

@@ -71,14 +71,14 @@ describe('DelegatedContentRouting', function () {
       expect(() => new DelegatedContentRouting()).to.throw()
     })
 
-    it('should default to https://ipfs.io as the delegate', () => {
+    it('should default to https://node0.delegate.ipfs.io as the delegate', () => {
       const router = new DelegatedContentRouting(selfId)
 
       expect(router.api).to.include({
         'api-path': '/api/v0/',
         protocol: 'https',
         port: 443,
-        host: 'ipfs.io'
+        host: 'node0.delegate.ipfs.io'
       })
     })