feat: reprovider

Author: vasco-santos

package.json

@@ -89,7 +89,7 @@
     "hapi-pino": "^6.0.0",
     "human-to-milliseconds": "^1.0.0",
     "interface-datastore": "~0.6.0",
-    "ipfs-bitswap": "~0.24.1",
+    "ipfs-bitswap": "ipfs/js-ipfs-bitswap#feat/use-ipfs-provider",
     "ipfs-block": "~0.8.1",
     "ipfs-block-service": "~0.15.1",
     "ipfs-http-client": "^32.0.0",

src/core/components/files-regular/refs-local-pull-stream.js

@@ -1,9 +1,8 @@
 'use strict'
 
-const CID = require('cids')
-const base32 = require('base32.js')
 const pull = require('pull-stream')
 const pullDefer = require('pull-defer')
+const { blockKeyToCid } = require('../../utils')
 
 module.exports = function (self) {
   return () => {
@@ -14,21 +13,12 @@ module.exports = function (self) {
         return deferred.resolve(pull.error(err))
       }
 
-      const refs = blocks.map(b => dsKeyToRef(b.key))
+      const refs = blocks.map(b => ({
+        ref: blockKeyToCid(b.key).toString()
+      }))
       deferred.resolve(pull.values(refs))
     })
 
     return deferred
   }
 }
-
-function dsKeyToRef (key) {
-  try {
-    // Block key is of the form /<base32 encoded string>
-    const decoder = new base32.Decoder()
-    const buff = Buffer.from(decoder.write(key.toString().slice(1)).finalize())
-    return { ref: new CID(buff).toString() }
-  } catch (err) {
-    return { err: `Could not convert block with key '${key}' to CID: ${err.message}` }
-  }
-}

src/core/components/start.js

@@ -1,11 +1,13 @@
 'use strict'
 
+const get = require('dlv')
 const series = require('async/series')
 const Bitswap = require('ipfs-bitswap')
 const setImmediate = require('async/setImmediate')
 const promisify = require('promisify-es6')
 
 const IPNS = require('../ipns')
+const Provider = require('../provider')
 const routingConfig = require('../ipns/routing/config')
 const createLibp2pBundle = require('./libp2p')
 
@@ -45,6 +47,8 @@ module.exports = (self) => {
           libp2p.start(err => {
             if (err) return cb(err)
             self.libp2p = libp2p
+
+            self._provider = new Provider(libp2p, self._repo.blocks, get(config, 'Reprovider'))
             cb()
           })
         })
@@ -56,9 +60,15 @@ module.exports = (self) => {
         self._bitswap = new Bitswap(
           self.libp2p,
           self._repo.blocks,
+          self._provider,
           { statsEnabled: true }
         )
 
+        if (!get(self._options, 'offline') &&
+          (get(self._options, 'libp2p.config.dht.enabled', false) || get(self._options, 'libp2p.modules.contentRouting', false))) {
+          self._provider.start()
+        }
+
         self._bitswap.start()
         self._blockService.setExchange(self._bitswap)
 

src/core/config.js

@@ -66,6 +66,10 @@ const configSchema = s({
         Enabled: 'boolean?'
       }))
     })),
+    Reprovider: optional(s({
+      Interval: 'string?',
+      Strategy: 'string?'
+    })),
     Bootstrap: optional(s(['multiaddr-ipfs']))
   })),
   ipld: 'object?',

src/core/provider/index.js

@@ -0,0 +1,93 @@
+'use strict'
+
+const errCode = require('err-code')
+const human = require('human-to-milliseconds')
+const promisify = require('promisify-es6')
+
+const CID = require('cids')
+
+const Reprovider = require('./reprovider')
+
+class Provider {
+  /**
+   * Provider goal is to announce blocks to the network.
+   * It keeps track of which blocks are provided, and allow them to be reprovided
+   * @param {Libp2p} libp2p
+   * @param {Blockstore} blockstore
+   * @param {object} options
+   * @memberof Provider
+   */
+  constructor (libp2p, blockstore, options) {
+    this._running = false
+
+    this._contentRouting = libp2p.contentRouting
+    this._blockstore = blockstore
+    this._options = options
+    this.reprovider = undefined
+  }
+
+  /**
+   * Begin processing the provider work
+   * @returns {void}
+   */
+  async start () {
+    // do not run twice
+    if (this._running) {
+      return
+    }
+
+    this._running = true
+
+    // handle options
+    const strategy = this._options.strategy || 'all'
+    const humanInterval = this._options.Interval || '12h'
+    const interval = await promisify((callback) => human(humanInterval, callback))()
+    const options = {
+      interval,
+      strategy
+    }
+
+    this.reprovider = new Reprovider(this._contentRouting, this._blockstore, options)
+
+    // Start reprovider
+    this.reprovider.start()
+  }
+
+  /**
+   * Stop the provider
+   * @returns {void}
+   */
+  stop () {
+    this._running = true
+
+    // stop the reprovider
+    this.reprovider.stop()
+  }
+
+  /**
+   * Announce block to the network and add and entry to the tracker
+   * Takes a cid and makes an attempt to announce it to the network
+   * @param {CID} cid
+   */
+  async provide (cid) {
+    if (!CID.isCID(cid)) {
+      throw errCode('invalid CID to provide', 'ERR_INVALID_CID')
+    }
+
+    await promisify((callback) => {
+      this._contentRouting.provide(cid, callback)
+    })()
+  }
+
+  async findProviders (cid, options) { // eslint-disable-line require-await
+    if (!CID.isCID(cid)) {
+      throw errCode('invalid CID to find', 'ERR_INVALID_CID')
+    }
+
+    return promisify((callback) => {
+      this._contentRouting.findProviders(cid, options, callback)
+    })()
+  }
+}
+
+exports = module.exports = Provider

src/core/provider/queue.js

@@ -0,0 +1,110 @@
+'use strict'
+
+const queue = require('async/queue')
+
+const debug = require('debug')
+const log = debug('ipfs:provider')
+log.error = debug('ipfs:provider:error')
+
+class WorkerQueue {
+  /**
+   * Creates an instance of a WorkerQueue.
+   * @param {function} executeWork
+   * @param {number} [concurrency=3]
+   */
+  constructor (executeWork, concurrency = 3) {
+    this.executeWork = executeWork
+    this._concurrency = concurrency
+
+    this.running = false
+    this.queue = this._setupQueue()
+  }
+
+  /**
+   * Create the underlying async queue.
+   * @returns {queue}
+   */
+  _setupQueue () {
+    const q = queue(async (block) => {
+      await this._processNext(block)
+    }, this._concurrency)
+
+    // If there is an error, stop the worker
+    q.error = (err) => {
+      log.error(err)
+      this.stop(err)
+    }
+
+    q.buffer = 0
+
+    return q
+  }
+
+  /**
+   * Use the queue from async to keep `concurrency` amount items running
+   * @param {Block[]} blocks
+   * @returns {Promise}
+   */
+  async execute (blocks) {
+    this.running = true
+
+    // store the promise resolution functions to be resolved at end of queue
+    this.execution = {}
+    const execPromise = new Promise((resolve, reject) => Object.assign(this.execution, { resolve, reject }))
+
+    // When all blocks have been processed, stop the worker
+    this.queue.drain = () => {
+      log('queue:drain')
+      this.stop()
+    }
+
+    // Fill queue with blocks
+    this.queue.push(blocks)
+
+    await execPromise
+  }
+
+  /**
+   * Stop the worker, optionally an error is thrown if received
+   *
+   * @param {object} error
+   */
+  stop (error) {
+    if (!this.running) {
+      return
+    }
+
+    this.running = false
+    this.queue.kill()
+
+    if (error) {
+      this.execution && this.execution.reject(error)
+    } else {
+      this.execution && this.execution.resolve()
+    }
+  }
+
+  /**
+   * Process the next block in the queue.
+   * @param {Block} block
+   */
+  async _processNext (block) {
+    if (!this.running) {
+      return
+    }
+
+    // Execute work
+    log('queue:work')
+
+    let execErr
+    try {
+      await this.executeWork(block)
+    } catch (err) {
+      execErr = err
+    }
+
+    log('queue:work:done', execErr)
+  }
+}
+
+exports = module.exports = WorkerQueue

src/core/provider/reprovider.js

@@ -0,0 +1,89 @@
+'use strict'
+
+const promisify = require('promisify-es6')
+const WorkerQueue = require('./queue')
+
+const { blockKeyToCid } = require('../utils')
+
+// const initialDelay = 15000
+const initialDelay = 3000
+
+class Reprovider {
+  /**
+   * Reprovider goal is to reannounce blocks to the network.
+   * @param {object} contentRouting
+   * @param {Blockstore} blockstore
+   * @param {object} options
+   * @memberof Reprovider
+   */
+  constructor (contentRouting, blockstore, options) {
+    this._contentRouting = contentRouting
+    this._blockstore = blockstore
+    this._options = options
+
+    this._timeoutId = undefined
+    this._worker = new WorkerQueue(this._provideBlock)
+  }
+
+  /**
+   * Begin processing the reprovider work and waiting for reprovide triggers
+   * @returns {void}
+   */
+  start () {
+    // Start doing reprovides after the initial delay
+    this._timeoutId = setTimeout(() => {
+      this._runPeriodically()
+    }, initialDelay)
+  }
+
+  /**
+   * Stops the reprovider. Any active reprovide actions should be aborted
+   * @returns {void}
+   */
+  stop () {
+    if (this._timeoutId) {
+      clearTimeout(this._timeoutId)
+      this._timeoutId = undefined
+    }
+    this._worker.stop()
+  }
+
+  /**
+   * Run reprovide on every `options.interval` ms
+   * @returns {void}
+   */
+  async _runPeriodically () {
+    while (this._timeoutId) {
+      const blocks = await promisify((callback) => this._blockstore.query({}, callback))()
+
+      // TODO strategy logic here
+      if (this._options.strategy === 'pinned') {
+
+      } else if (this._options.strategy === 'pinned') {
+
+      }
+
+      await this._worker.execute(blocks)
+
+      // Each subsequent walk should run on a `this._options.interval` interval
+      await new Promise(resolve => {
+        this._timeoutId = setTimeout(resolve, this._options.interval)
+      })
+    }
+  }
+
+  /**
+   * Do the reprovide work to libp2p content routing
+   * @param {Block} block
+   * @returns {void}
+   */
+  async _provideBlock (block) {
+    const cid = blockKeyToCid(block.key.toBuffer())
+
+    await promisify((callback) => {
+      this._contentRouting.provide(cid, callback)
+    })()
+  }
+}
+
+exports = module.exports = Reprovider