feat: allow passing a http.Agent to ipfs-http-client in node (#3474)

Right now no `http.Agent` is used for requests made using the http client in node, which means each request opens a new connection which can end up hitting process resource limits which means connections get dropped.

The change here sets a default `http.Agent` with a `keepAlive: true` and `maxSockets` of 6 which is consistent with [browsers](https://tools.ietf.org/html/rfc2616#section-8.1.4) and [native apps](https://developer.apple.com/documentation/foundation/nsurlsessionconfiguration/1407597-httpmaximumconnectionsperhost?language=objc).

The user can override the agent passed to the `ipfs-http-client` constructor to restore the previous functionality:

```js
const http = require('http')
const createClient = require('ipfs-http-client')

const client = createClient({
  url: 'http://127.0.0.1:5002',
  agent: new http.Agent({
    keepAlive: false,
    maxSockets: Infinity
  })
})
```

Refs: #3464

Author: achingbrain

README.md

@@ -99,6 +99,7 @@ All core API methods take _additional_ `options` specific to the HTTP API:
 
 * `headers` - An object or [Headers](https://developer.mozilla.org/en-US/docs/Web/API/Headers) instance that can be used to set custom HTTP headers. Note that this option can also be [configured globally](#custom-headers) via the constructor options.
 * `searchParams` - An object or [`URLSearchParams`](https://developer.mozilla.org/en-US/docs/Web/API/URLSearchParams) instance that can be used to add additional query parameters to the query string sent with each request.
+* `agent` - A node [http.Agent](https://nodejs.org/api/http.html#http_class_http_agent) used to configure connection persistence and reuse (only supported in node.js)
 
 ### Instance Utils
 

package.json

@@ -18,7 +18,8 @@
     "./src/lib/multipart-request.js": "./src/lib/multipart-request.browser.js",
     "ipfs-utils/src/files/glob-source": false,
     "go-ipfs": false,
-    "ipfs-core-utils/src/files/normalise-input": "ipfs-core-utils/src/files/normalise-input/index.browser.js"
+    "ipfs-core-utils/src/files/normalise-input": "ipfs-core-utils/src/files/normalise-input/index.browser.js",
+    "http": false
   },
   "typesVersions": {
     "*": {
@@ -79,6 +80,7 @@
   },
   "devDependencies": {
     "aegir": "^29.2.2",
+    "delay": "^4.4.0",
     "go-ipfs": "^0.7.0",
     "ipfs-core": "^0.3.1",
     "ipfsd-ctl": "^7.2.0",

src/lib/core.js

@@ -1,12 +1,13 @@
 'use strict'
 /* eslint-env browser */
 const Multiaddr = require('multiaddr')
-const { isBrowser, isWebWorker } = require('ipfs-utils/src/env')
+const { isBrowser, isWebWorker, isNode } = require('ipfs-utils/src/env')
 const parseDuration = require('parse-duration').default
 const log = require('debug')('ipfs-http-client:lib:error-handler')
 const HTTP = require('ipfs-utils/src/http')
 const merge = require('merge-options')
 const toUrlString = require('ipfs-core-utils/src/to-url-string')
+const http = require('http')
 
 const DEFAULT_PROTOCOL = isBrowser || isWebWorker ? location.protocol : 'http'
 const DEFAULT_HOST = isBrowser || isWebWorker ? location.hostname : 'localhost'
@@ -19,6 +20,7 @@ const DEFAULT_PORT = isBrowser || isWebWorker ? location.port : '5001'
 const normalizeOptions = (options = {}) => {
   let url
   let opts = {}
+  let agent
 
   if (typeof options === 'string' || Multiaddr.isMultiaddr(options)) {
     url = new URL(toUrlString(options))
@@ -46,13 +48,22 @@ const normalizeOptions = (options = {}) => {
     url.pathname = 'api/v0'
   }
 
+  if (isNode) {
+    agent = opts.agent || new http.Agent({
+      keepAlive: true,
+      // Similar to browsers which limit connections to six per host
+      maxSockets: 6
+    })
+  }
+
   return {
     ...opts,
     host: url.host,
     protocol: url.protocol.replace(':', ''),
     port: Number(url.port),
     apiPath: url.pathname,
-    url
+    url,
+    agent
   }
 }
 
@@ -105,6 +116,8 @@ const parseTimeout = (value) => {
 }
 
 /**
+ * @typedef {import('http').Agent} Agent
+ *
  * @typedef {Object} ClientOptions
  * @property {string} [host]
  * @property {number} [port]
@@ -116,6 +129,7 @@ const parseTimeout = (value) => {
  * @property {object} [ipld]
  * @property {any[]} [ipld.formats] - An array of additional [IPLD formats](https://github.com/ipld/interface-ipld-format) to support
  * @property {(format: string) => Promise<any>} [ipld.loadFormat] - an async function that takes the name of an [IPLD format](https://github.com/ipld/interface-ipld-format) as a string and should return the implementation of that codec
+ * @property {Agent} [agent] - A [http.Agent](https://nodejs.org/api/http.html#http_class_http_agent) used to control connection persistence and reuse for HTTP clients (only supported in node.js)
  */
 class Client extends HTTP {
   /**
@@ -149,7 +163,8 @@ class Client extends HTTP {
         }
 
         return out
-      }
+      },
+      agent: opts.agent
     })
 
     delete this.get

test/node.js

@@ -1,5 +1,6 @@
 'use strict'
 
+require('./node/agent')
 require('./node/swarm')
 require('./node/request-api')
 require('./node/custom-headers')

test/node/agent.js

@@ -0,0 +1,111 @@
+/* eslint-env mocha */
+'use strict'
+
+const { expect } = require('aegir/utils/chai')
+const ipfsClient = require('../../src')
+const delay = require('delay')
+
+function startServer (handler) {
+  return new Promise((resolve) => {
+    // spin up a test http server to inspect the requests made by the library
+    const server = require('http').createServer((req, res) => {
+      req.on('data', () => {})
+      req.on('end', async () => {
+        const out = await handler(req)
+
+        res.writeHead(200)
+        res.write(JSON.stringify(out))
+        res.end()
+      })
+    })
+
+    server.listen(0, () => {
+      resolve({
+        port: server.address().port,
+        close: () => server.close()
+      })
+    })
+  })
+}
+
+describe('agent', function () {
+  let agent
+
+  before(() => {
+    const { Agent } = require('http')
+
+    agent = new Agent({
+      maxSockets: 2
+    })
+  })
+
+  it('restricts the number of concurrent connections', async () => {
+    const responses = []
+
+    const server = await startServer(() => {
+      const p = new Promise((resolve) => {
+        responses.push(resolve)
+      })
+
+      return p
+    })
+
+    const ipfs = ipfsClient({
+      url: `http://localhost:${server.port}`,
+      agent
+    })
+
+    // make three requests
+    const requests = Promise.all([
+      ipfs.id(),
+      ipfs.id(),
+      ipfs.id()
+    ])
+
+    // wait for the first two to arrive
+    for (let i = 0; i < 5; i++) {
+      await delay(100)
+
+      if (responses.length === 2) {
+        // wait a little longer, the third should not arrive
+        await delay(1000)
+
+        expect(responses).to.have.lengthOf(2)
+
+        // respond to the in-flight requests
+        responses[0]({
+          res: 0
+        })
+        responses[1]({
+          res: 1
+        })
+
+        break
+      }
+    }
+
+    // wait for the final request to arrive
+    for (let i = 0; i < 5; i++) {
+      await delay(100)
+
+      if (responses.length === 3) {
+        // respond to it
+        responses[2]({
+          res: 2
+        })
+      }
+    }
+
+    const results = await requests
+
+    expect(results).to.deep.equal([{
+      res: 0
+    }, {
+      res: 1
+    }, {
+      res: 2
+    }])
+
+    server.close()
+  })
+})