refactor: async await

package.json

@@ -32,42 +32,46 @@
  "url": "https://github.com/libp2p/js-libp2p-kad-dht/issues"
  },
  "engines": {
- "node": ">=6.0.0",
- "npm": ">=3.0.0"
+ "node": ">=10.0.0",
+ "npm": ">=6.0.0"
  },
  "homepage": "https://github.com/libp2p/js-libp2p-kad-dht",
  "dependencies": {
  "abort-controller": "^3.0.0",
  "async": "^2.6.2",
  "base32.js": "~0.1.0",
  "chai-checkmark": "^1.0.1",
- "cids": "~0.7.0",
+ "cids": "~0.7.1",
  "debug": "^4.1.1",
- "err-code": "^1.1.2",
+ "err-code": "^2.0.0",
  "hashlru": "^2.3.0",
  "heap": "~0.2.6",
- "interface-datastore": "~0.7.0",
+ "interface-datastore": "~0.8.0",
  "k-bucket": "^5.0.0",
- "libp2p-crypto": "~0.16.1",
- "libp2p-record": "~0.6.2",
- "multihashes": "~0.4.14",
- "multihashing-async": "~0.5.2",
- "p-queue": "^6.0.0",
+ "libp2p-crypto": "~0.17.1",
+ "libp2p-record": "~0.7.0",
+ "multihashes": "~0.4.15",
+ "multihashing-async": "~0.8.0",
+ "p-filter": "^2.1.0",
+ "p-map": "^3.0.0",
+ "p-queue": "^6.2.1",
+ "p-timeout": "^3.2.0",
  "p-times": "^2.1.0",
- "peer-id": "~0.12.2",
- "peer-info": "~0.15.1",
+ "peer-id": "~0.13.5",
+ "peer-info": "~0.17.0",
  "promise-to-callback": "^1.0.0",
  "promisify-es6": "^1.0.3",
  "protons": "^1.0.1",
- "pull-length-prefixed": "^1.3.2",
- "pull-stream": "^3.6.9",
+ "pull-length-prefixed": "^1.3.3",
+ "pull-stream": "^3.6.14",
  "varint": "^5.0.0",
  "xor-distance": "^2.0.0"
  },
  "devDependencies": {
- "aegir": "^20.0.0",
+ "aegir": "^20.4.1",
  "chai": "^4.2.0",
  "datastore-level": "~0.12.1",
+ "delay": "^4.3.0",
  "dirty-chai": "^2.0.1",
  "interface-connection": "~0.3.3",
  "libp2p-mplex": "~0.8.5",
@@ -76,8 +80,12 @@
  "lodash": "^4.17.11",
  "lodash.random": "^3.2.0",
  "lodash.range": "^3.2.0",
- "peer-book": "~0.9.1",
- "sinon": "^7.3.1"
+ "p-defer": "^3.0.0",
+ "p-each-series": "^2.1.0",
+ "p-map-series": "^2.1.0",
+ "p-retry": "^4.2.0",
+ "peer-book": "~0.9.2",
+ "sinon": "^7.5.0"
  },
  "contributors": [
  "Alan Shaw <alan.shaw@protocol.ai>",

src/content-fetching/index.js

@@ -0,0 +1,278 @@
+'use strict'
+
+const errcode = require('err-code')
+
+const pFilter = require('p-filter')
+const pTimeout = require('p-timeout')
+
+const libp2pRecord = require('libp2p-record')
+
+const c = require('../constants')
+const Query = require('../query')
+
+const utils = require('../utils')
+
+const Record = libp2pRecord.Record
+
+module.exports = (dht) => {
+ const putLocal = async (key, rec) => { // eslint-disable-line require-await
+ return dht.datastore.put(utils.bufferToKey(key), rec)
+ }
+
+ /**
+ * Attempt to retrieve the value for the given key from
+ * the local datastore.
+ *
+ * @param {Buffer} key
+ * @returns {Promise<Record>}
+ *
+ * @private
+ */
+ const getLocal = async (key) => {
+ dht._log('getLocal %b', key)
+
+ const raw = await dht.datastore.get(utils.bufferToKey(key))
+ dht._log('found %b in local datastore', key)
+ const rec = Record.deserialize(raw)
+
+ await dht._verifyRecordLocally(rec)
+ return rec
+ }
+
+ /**
+ * Send the best record found to any peers that have an out of date record.
+ *
+ * @param {Buffer} key
+ * @param {Array<Object>} vals - values retrieved from the DHT
+ * @param {Object} best - the best record that was found
+ * @returns {Promise}
+ *
+ * @private
+ */
+ const sendCorrectionRecord = async (key, vals, best) => {
+ const fixupRec = await utils.createPutRecord(key, best)
+
+ return Promise.all(vals.map(async (v) => {
+ // no need to do anything
+ if (v.val.equals(best)) {
+ return
+ }
+
+ // correct ourself
+ if (dht._isSelf(v.from)) {
+ try {
+ await dht._putLocal(key, fixupRec)
+ } catch (err) {
+ dht._log.error('Failed error correcting self', err)
+ }
+ return
+ }
+
+ // send correction
+ try {
+ await dht._putValueToPeer(key, fixupRec, v.from)
+ } catch (err) {
+ dht._log.error('Failed error correcting entry', err)
+ }
+ }))
+ }
+
+ return {
+ /**
+ * Store the given key/value pair locally, in the datastore.
+ * @param {Buffer} key
+ * @param {Buffer} rec - encoded record
+ * @returns {Promise<void>}
+ * @private
+ */
+ async _putLocal (key, rec) { // eslint-disable-line require-await
+ return putLocal(key, rec)
+ },
+
+ /**
+ * Store the given key/value pair in the DHT.
+ *
+ * @param {Buffer} key
+ * @param {Buffer} value
+ * @param {Object} [options] - put options
+ * @param {number} [options.minPeers] - minimum number of peers required to successfully put (default: closestPeers.length)
+ * @returns {Promise<void>}
+ */
+ async put (key, value, options = {}) {
+ dht._log('PutValue %b', key)
+
+ // create record in the dht format
+ const record = await utils.createPutRecord(key, value)
+
+ // store the record locally
+ await putLocal(key, record)
+
+ // put record to the closest peers
+ const peers = await dht.getClosestPeers(key, { shallow: true })
+ const results = await pFilter(peers, async (peer) => {
+ try {
+ await dht._putValueToPeer(key, record, peer)
+ return true
+ } catch (err) {
+ dht._log.error('Failed to put to peer (%b): %s', peer.id, err)
+ return false
+ }
+ })
+
+ // verify if we were able to put to enough peers
+ const minPeers = options.minPeers || peers.length // Ensure we have a default `minPeers`
+
+ if (minPeers > results.length) {
+ const error = errcode(new Error(`Failed to put value to enough peers: ${results.length}/${minPeers}`), 'ERR_NOT_ENOUGH_PUT_PEERS')
+ dht._log.error(error)
+ throw error
+ }
+ },
+
+ /**
+ * Get the value to the given key.
+ * Times out after 1 minute by default.
+ *
+ * @param {Buffer} key
+ * @param {Object} [options] - get options
+ * @param {number} [options.timeout] - optional timeout (default: 60000)
+ * @returns {Promise<{from: PeerId, val: Buffer}>}
+ */
+ async get (key, options = {}) {
+ options.timeout = options.timeout || c.minute
+
+ dht._log('_get %b', key)
+
+ const vals = await dht.getMany(key, c.GET_MANY_RECORD_COUNT, options)
+ const recs = vals.map((v) => v.val)
+ let i = 0
+
+ try {
+ i = libp2pRecord.selection.bestRecord(dht.selectors, key, recs)
+ } catch (err) {
+ // Assume the first record if no selector available
+ if (err.code !== 'ERR_NO_SELECTOR_FUNCTION_FOR_RECORD_KEY') {
+ throw err
+ }
+ }
+
+ const best = recs[i]
+ dht._log('GetValue %b %s', key, best)
+
+ if (!best) {
+ throw errcode(new Error('best value was not found'), 'ERR_NOT_FOUND')
+ }
+
+ await sendCorrectionRecord(key, vals, best)
+
+ return best
+ },
+
+ /**
+ * Get the `n` values to the given key without sorting.
+ *
+ * @param {Buffer} key
+ * @param {number} nvals
+ * @param {Object} [options] - get options
+ * @param {number} [options.timeout] - optional timeout (default: 60000)
+ * @returns {Promise<Array<{from: PeerId, val: Buffer}>>}
+ */
+ async getMany (key, nvals, options = {}) {
+ options.timeout = options.timeout || c.minute
+
+ dht._log('getMany %b (%s)', key, nvals)
+
+ let vals = []
+ let localRec
+
+ try {
+ localRec = await getLocal(key)
+ } catch (err) {
+ if (nvals === 0) {
+ throw err
+ }
+ }
+
+ if (localRec) {
+ vals.push({
+ val: localRec.value,
+ from: dht.peerInfo.id
+ })
+ }
+
+ if (vals.length >= nvals) {
+ return vals
+ }
+
+ const paths = []
+ const id = await utils.convertBuffer(key)
+ const rtp = dht.routingTable.closestPeers(id, this.kBucketSize)
+
+ dht._log('peers in rt: %d', rtp.length)
+
+ if (rtp.length === 0) {
+ const errMsg = 'Failed to lookup key! No peers from routing table!'
+
+ dht._log.error(errMsg)
+ throw errcode(new Error(errMsg), 'ERR_NO_PEERS_IN_ROUTING_TABLE')
+ }
+
+ // we have peers, lets do the actual query to them
+ const query = new Query(dht, key, (pathIndex, numPaths) => {
+ // This function body runs once per disjoint path
+ const pathSize = utils.pathSize(nvals - vals.length, numPaths)
+ const pathVals = []
+ paths.push(pathVals)
+
+ // Here we return the query function to use on this particular disjoint path
+ return async (peer) => {
+ let rec, peers, lookupErr
+ try {
+ const results = await dht._getValueOrPeers(peer, key)
+ rec = results.record
+ peers = results.peers
+ } catch (err) {
+ // If we have an invalid record we just want to continue and fetch a new one.
+ if (err.code !== 'ERR_INVALID_RECORD') {
+ throw err
+ }
+ lookupErr = err
+ }
+
+ const res = { closerPeers: peers }
+
+ if ((rec && rec.value) || lookupErr) {
+ pathVals.push({
+ val: rec && rec.value,
+ from: peer
+ })
+ }
+
+ // enough is enough
+ if (pathVals.length >= pathSize) {
+ res.pathComplete = true
+ }
+
+ return res
+ }
+ })
+
+ let error
+ try {
+ await pTimeout(query.run(rtp), options.timeout)
+ } catch (err) {
+ error = err
+ }
+ query.stop()
+
+ // combine vals from each path
+ vals = [].concat.apply(vals, paths).slice(0, nvals)
+
+ if (error && vals.length === 0) {
+ throw error
+ }
+
+ return vals
+ }
+ }
+}