refactor: convert preload to async/await (#2590)

...and removes `async` from dependencies!

And half the code.

License: MIT
Signed-off-by: Alan Shaw <[email protected]>

Author: Alan Shaw

.aegir.js

@@ -4,6 +4,7 @@ const IPFSFactory = require('ipfsd-ctl')
 const parallel = require('async/parallel')
 const MockPreloadNode = require('./test/utils/mock-preload-node')
 const EchoServer = require('interface-ipfs-core/src/utils/echo-http-server')
+const callbackify = require('callbackify')
 
 const ipfsdServer = IPFSFactory.createServer()
 const preloadNode = MockPreloadNode.createNode()
@@ -30,13 +31,13 @@ module.exports = {
     node: {
       pre: (cb) => {
         parallel([
-          (cb) => preloadNode.start(cb),
+          (cb) => callbackify(preloadNode.start)(cb),
           (cb) => echoServer.start(cb)
         ], cb)
       },
       post: (cb) => {
         parallel([
-          (cb) => preloadNode.stop(cb),
+          (cb) => callbackify(preloadNode.stop)(cb),
           (cb) => echoServer.stop(cb)
         ], cb)
       }
@@ -48,7 +49,7 @@ module.exports = {
             ipfsdServer.start()
             cb()
           },
-          (cb) => preloadNode.start(cb),
+          (cb) => callbackify(preloadNode.start)(cb),
           (cb) => echoServer.start(cb)
         ], cb)
       },
@@ -58,7 +59,7 @@ module.exports = {
             ipfsdServer.stop()
             cb()
           },
-          (cb) => preloadNode.stop(cb),
+          (cb) => callbackify(preloadNode.stop)(cb),
           (cb) => echoServer.stop(cb)
         ], cb)
       }

package.json

@@ -64,8 +64,8 @@
     "@hapi/boom": "^7.4.3",
     "@hapi/hapi": "^18.3.2",
     "@hapi/joi": "^15.0.0",
+    "abort-controller": "^3.0.0",
     "array-shuffle": "^1.0.1",
-    "async": "^2.6.1",
     "async-iterator-all": "^1.0.0",
     "async-iterator-first": "^1.0.0",
     "async-iterator-to-pull-stream": "^1.3.0",
@@ -146,7 +146,6 @@
     "libp2p-webrtc-star": "~0.16.0",
     "libp2p-websocket-star-multi": "~0.4.3",
     "libp2p-websockets": "~0.12.3",
-    "lodash": "^4.17.15",
     "lodash.flatten": "^4.4.0",
     "mafmt": "^6.0.10",
     "merge-options": "^2.0.0",
@@ -196,6 +195,7 @@
   },
   "devDependencies": {
     "aegir": "^20.4.1",
+    "async": "^2.6.3",
     "base64url": "^3.0.1",
     "clear-module": "^4.0.0",
     "delay": "^4.1.0",
@@ -208,6 +208,7 @@
     "ipfs-interop": "^0.1.1",
     "ipfsd-ctl": "^0.47.2",
     "libp2p-websocket-star": "~0.10.2",
+    "lodash": "^4.17.15",
     "ncp": "^2.0.0",
     "p-event": "^4.1.0",
     "qs": "^6.5.2",

src/core/components/init-assets.js

@@ -8,7 +8,8 @@ module.exports = async function addDefaultAssets (self, log) {
   const initDocsPath = path.join(__dirname, '../../init-files/init-docs')
 
   const results = await self.addFromFs(initDocsPath, {
-    recursive: true
+    recursive: true,
+    preload: false
   })
 
   const dir = results.filter(file => file.path === 'init-docs').pop()

src/core/components/init.js

@@ -116,7 +116,8 @@ async function addRepoAssets (self, privateKey, opts) {
   const cid = await self.dag.put(node, {
     version: 0,
     format: multicodec.DAG_PB,
-    hashAlg: multicodec.SHA2_256
+    hashAlg: multicodec.SHA2_256,
+    preload: false
   })
 
   await self._ipns.initializeKeyspace(privateKey, cid.toBaseEncodedString())

src/core/mfs-preload.js

@@ -19,24 +19,20 @@ module.exports = (self) => {
   let rootCid
   let timeoutId
 
-  const preloadMfs = () => {
-    self.files.stat('/')
-      .then((stats) => {
-        if (rootCid !== stats.hash) {
-          log(`preloading updated MFS root ${rootCid} -> ${stats.hash}`)
-
-          return self._preload(stats.hash, (err) => {
-            timeoutId = setTimeout(preloadMfs, options.interval)
-            if (err) return log.error(`failed to preload MFS root ${stats.hash}`, err)
-            rootCid = stats.hash
-          })
-        }
+  const preloadMfs = async () => {
+    try {
+      const stats = await self.files.stat('/')
 
-        timeoutId = setTimeout(preloadMfs, options.interval)
-      }, (err) => {
-        timeoutId = setTimeout(preloadMfs, options.interval)
-        log.error('failed to stat MFS root for preload', err)
-      })
+      if (rootCid !== stats.hash) {
+        log(`preloading updated MFS root ${rootCid} -> ${stats.hash}`)
+        await self._preload(stats.hash)
+        rootCid = stats.hash
+      }
+    } catch (err) {
+      log.error('failed to preload MFS root', err)
+    } finally {
+      timeoutId = setTimeout(preloadMfs, options.interval)
+    }
   }
 
   return {

src/core/preload.js

@@ -1,30 +1,23 @@
 'use strict'
 
-const setImmediate = require('async/setImmediate')
-const retry = require('async/retry')
 const toUri = require('multiaddr-to-uri')
 const debug = require('debug')
 const CID = require('cids')
 const shuffle = require('array-shuffle')
+const AbortController = require('abort-controller')
 const preload = require('./runtime/preload-nodejs')
 
 const log = debug('ipfs:preload')
 log.error = debug('ipfs:preload:error')
 
-const noop = (err) => { if (err) log.error(err) }
-
 module.exports = self => {
   const options = self._options.preload || {}
   options.enabled = Boolean(options.enabled)
   options.addresses = options.addresses || []
 
   if (!options.enabled || !options.addresses.length) {
     log('preload disabled')
-    const api = (_, callback) => {
-      if (callback) {
-        setImmediate(() => callback())
-      }
-    }
+    const api = () => {}
     api.start = () => {}
     api.stop = () => {}
     return api
@@ -34,41 +27,40 @@ module.exports = self => {
   let requests = []
   const apiUris = options.addresses.map(toUri)
 
-  const api = (path, callback) => {
-    callback = callback || noop
+  const api = async path => {
+    try {
+      if (stopped) throw new Error(`preload ${path} but preloader is not started`)
 
-    if (typeof path !== 'string') {
-      try {
-        path = new CID(path).toBaseEncodedString()
-      } catch (err) {
-        return setImmediate(() => callback(err))
+      if (typeof path !== 'string') {
+        path = new CID(path).toString()
       }
-    }
-
-    const fallbackApiUris = shuffle(apiUris)
-    let request
-    const now = Date.now()
 
-    retry({ times: fallbackApiUris.length }, (cb) => {
-      if (stopped) return cb(new Error(`preload aborted for ${path}`))
-
-      // Remove failed request from a previous attempt
-      requests = requests.filter(r => r !== request)
-
-      const apiUri = fallbackApiUris.shift()
-
-      request = preload(`${apiUri}/api/v0/refs?r=true&arg=${encodeURIComponent(path)}`, cb)
-      requests = requests.concat(request)
-    }, (err) => {
-      requests = requests.filter(r => r !== request)
-
-      if (err) {
-        return callback(err)
+      const fallbackApiUris = shuffle(apiUris)
+      let success = false
+      const now = Date.now()
+
+      for (const uri of fallbackApiUris) {
+        if (stopped) throw new Error(`preload aborted for ${path}`)
+        let controller
+
+        try {
+          controller = new AbortController()
+          requests = requests.concat(controller)
+          await preload(`${uri}/api/v0/refs?r=true&arg=${encodeURIComponent(path)}`, { signal: controller.signal })
+          success = true
+        } catch (err) {
+          if (err.type !== 'aborted') log.error(err)
+        } finally {
+          requests = requests.filter(r => r !== controller)
+        }
+
+        if (success) break
       }
 
-      log(`preloaded ${path} in ${Date.now() - now}ms`)
-      callback()
-    })
+      log(`${success ? '' : 'un'}successfully preloaded ${path} in ${Date.now() - now}ms`)
+    } catch (err) {
+      log.error(err)
+    }
   }
 
   api.start = () => {
@@ -77,8 +69,8 @@ module.exports = self => {
 
   api.stop = () => {
     stopped = true
-    log(`canceling ${requests.length} pending preload request(s)`)
-    requests.forEach(r => r.cancel())
+    log(`aborting ${requests.length} pending preload request(s)`)
+    requests.forEach(r => r.abort())
     requests = []
   }
 

src/core/runtime/preload-browser.js

@@ -12,16 +12,22 @@ log.error = debug('ipfs:preload:error')
 // we don't want preload calls to exhaust the limit (~6)
 const httpQueue = new PQueue({ concurrency: 4 })
 
-module.exports = function preload (url, callback) {
+module.exports = function preload (url, options) {
   log(url)
+  options = options || {}
 
-  const controller = new AbortController()
-  const signal = controller.signal
-  const cb = () => setImmediate(callback) // https://github.com/ipfs/js-ipfs/pull/2304#discussion_r320700893
+  return httpQueue.add(async () => {
+    const res = await ky.get(url, { signal: options.signal })
+    const reader = res.body.getReader()
 
-  httpQueue.add(() => ky.get(url, { signal })).then(cb, cb)
-
-  return {
-    cancel: () => controller.abort()
-  }
+    try {
+      while (true) {
+        const { done } = await reader.read()
+        if (done) return
+        // Read to completion but do not cache
+      }
+    } finally {
+      reader.releaseLock()
+    }
+  })
 }

src/core/runtime/preload-nodejs.js

@@ -1,64 +1,18 @@
 'use strict'
 
-const http = require('http')
-const https = require('https')
-const { URL } = require('url')
+const { default: ky } = require('ky-universal')
 const debug = require('debug')
-const setImmediate = require('async/setImmediate')
 
 const log = debug('ipfs:preload')
 log.error = debug('ipfs:preload:error')
 
-module.exports = function preload (url, callback = () => {}) {
+module.exports = async function preload (url, options) {
   log(url)
+  options = options || {}
 
-  try {
-    url = new URL(url)
-  } catch (err) {
-    return setImmediate(() => callback(err))
-  }
-
-  const transport = url.protocol === 'https:' ? https : http
-
-  const req = transport.get({
-    hostname: url.hostname,
-    port: url.port,
-    path: url.pathname + url.search
-  }, (res) => {
-    if (res.statusCode < 200 || res.statusCode >= 300) {
-      res.resume()
-      log.error('failed to preload', url.href, res.statusCode, res.statusMessage)
-      return callback(new Error(`failed to preload ${url}`))
-    }
-
-    res.on('data', chunk => log(`data ${chunk}`))
-
-    res.on('abort', () => {
-      callback(new Error('request aborted'))
-    })
-
-    res.on('error', err => {
-      log.error('response error preloading', url.href, err)
-      callback(err)
-    })
-
-    res.on('end', () => {
-      // If aborted, callback is called in the abort handler
-      if (!res.aborted) callback()
-    })
-  })
-
-  req.on('error', err => {
-    log.error('request error preloading', url.href, err)
-    callback(err)
-  })
+  const res = await ky.get(url, { signal: options.signal })
 
-  return {
-    cancel: () => {
-      // No need to call callback here
-      // before repsonse - called in req error handler
-      // after response - called in res abort hander
-      req.abort()
-    }
+  for await (const _ of res.body) { // eslint-disable-line no-unused-vars
+    // Read to completion but do not cache
   }
 }

test/core/mfs-preload.spec.js

@@ -15,11 +15,7 @@ const createMockFilesStat = (cids = []) => {
 }
 
 const createMockPreload = () => {
-  const preload = (cid, cb = () => {}) => {
-    preload.cids.push(cid)
-
-    cb()
-  }
+  const preload = cid => preload.cids.push(cid)
   preload.cids = []
   return preload
 }