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,11 @@ module.exports = (self) => {
         self._bitswap = new Bitswap(
           self.libp2p,
           self._repo.blocks,
+          self._provider,
           { statsEnabled: true }
         )
 
+        self._provider.start()
         self._bitswap.start()
         self._blockService.setExchange(self._bitswap)
 

src/core/provider/index.js

@@ -0,0 +1,83 @@
+'use strict'
+
+const errCode = require('err-code')
+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) {
+    // the CIDs for which provide announcements should be made
+    // this.tracker = [] // TODO queue
+    this._running = false
+
+    this._contentRouting = libp2p.contentRouting
+    this._blockstore = blockstore
+
+    // TODO Convert
+    console.log('opts', options)
+
+    this.reprovider = new Reprovider(this._contentRouting, this._blockstore, options)
+  }
+
+  /**
+   * Begin processing the provider work
+   */
+  start () {
+    // do not run twice
+    if (this._running) {
+      return
+    }
+
+    this._running = true
+
+    // Start reprovider
+    this.reprovider.start()
+  }
+
+  /**
+   * Stop the provider
+   */
+  stop () {
+    this._running = true
+
+    // stop the reprovider
+    return 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
+   */
+  async provide (cid) {
+    if (!CID.isCID(key)) {
+      throw errCode('invalid CID to provide', 'ERR_INVALID_CID')
+    }
+
+    await promisify((callback) => {
+      this._contentRouting.provide(cid, callback)
+    })()
+  }
+
+  async findProviders (cid, options) {
+    if (!CID.isCID(key)) {
+      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,93 @@
+'use strict'
+
+const promisify = require('promisify-es6')
+
+const { blockKeyToCid } = require('../utils')
+
+// TODO Provide queue?
+
+// const initialDelay = 10000
+const initialDelay = 5000
+
+class Reprovider {
+  /**
+   * Reprovider goal is t§o reannounce blocks to the network.
+   * @param {*} contentRouting
+   * @param {Blockstore} blockstore
+   * @param {object} options
+   * @memberof Reprovider
+   */
+  constructor (contentRouting, blockstore, options) {
+    this._contentRouting = contentRouting
+    this._blockstore = blockstore // use query({})
+    this._options = options
+
+    this._timeoutId = undefined
+  }
+
+  /**
+   * Begin processing the reprovider work and waiting for reprovide triggers
+   * @returns {void}
+   */
+  start () {    
+    // Start doing reprovides after the initial delay
+    this._timeoutId = setTimeout(() => {
+      // start runner immediately
+      this._runPeriodically()
+    }, initialDelay)
+  }
+
+  /**
+   * Stops the reprovider. Any active reprovide actions should be aborted
+   */
+  stop() {
+    if (this._timeoutId) {
+      clearTimeout(this._timeoutId)
+      this._timeoutId = undefined
+    }
+  }
+
+  /**
+   * Run reprovide on every `options.interval` ms
+   * @returns {void}
+   */
+  async _runPeriodically () {
+    while (this._timeoutId) {
+      const blocks = await promisify((callback) => this._blockstore.query({}, callback))()
+
+      await this._reprovide(blocks)
+
+      await new Promise(resolve => {
+        this._timeoutId = setTimeout(resolve, 20000)
+      })
+
+      /*
+      // Each subsequent walk should run on a `this._options.interval` interval
+      await new Promise(resolve => {
+        this._timeoutId = setTimeout(resolve, this._options.interval)
+      }) */
+    }
+  }
+
+  /**
+   * provide all keys to libp2p content routing
+   */
+  async _reprovide (blocks) {
+    for (let i = 0; i < blocks.length && this._timeoutId; i++) {
+      const cid = blockKeyToCid(blocks[i].key.toBuffer())
+
+      console.log('cid')
+      // TODO: needs the DHT
+      // await promisify((callback) => {
+      //   this._contentRouting.provide(cid, callback)
+      // })()
+    }
+    console.log('end')
+  }
+
+  _dsKeyToCid (key) {
+
+  }
+}
+
+exports = module.exports = Reprovider

src/core/provider/reprovider.js

@@ -26,6 +26,10 @@ module.exports = () => ({
     '/dns4/node0.preload.ipfs.io/tcp/443/wss/ipfs/QmZMxNdpMkewiVZLMRxaNxUeZpDUb34pWjZ1kZvsd16Zic',
     '/dns4/node1.preload.ipfs.io/tcp/443/wss/ipfs/Qmbut9Ywz9YEDrz8ySBSgWyJk41Uvm2QJPhwDJzJyGFsD6'
   ],
+  Reprovider: {
+    Interval: '12h',
+    Strategy: 'all'
+  },
   Swarm: {
     ConnMgr: {
       LowWater: 200,

src/core/runtime/config-browser.js

@@ -39,6 +39,10 @@ module.exports = () => ({
     '/dns4/node0.preload.ipfs.io/tcp/443/wss/ipfs/QmZMxNdpMkewiVZLMRxaNxUeZpDUb34pWjZ1kZvsd16Zic',
     '/dns4/node1.preload.ipfs.io/tcp/443/wss/ipfs/Qmbut9Ywz9YEDrz8ySBSgWyJk41Uvm2QJPhwDJzJyGFsD6'
   ],
+  Reprovider: {
+    Interval: '12h',
+    Strategy: 'all'
+  },
   Swarm: {
     ConnMgr: {
       LowWater: 200,

src/core/runtime/config-nodejs.js

@@ -4,6 +4,7 @@ const promisify = require('promisify-es6')
 const map = require('async/map')
 const isIpfs = require('is-ipfs')
 const CID = require('cids')
+const base32 = require('base32.js')
 
 const ERR_BAD_PATH = 'ERR_BAD_PATH'
 exports.OFFLINE_ERROR = 'This command must be run in online mode. Try running \'ipfs daemon\' first.'
@@ -134,7 +135,22 @@ const resolvePath = promisify(function (objectAPI, ipfsPaths, callback) {
     }
   }, callback)
 })
+/**
+ * Convert a block key to cid
+ * @param {Key} key form /<base32 encoded string>
+ * @returns {CID}
+ */
+function blockKeyToCid (key) {
+  try {
+    const decoder = new base32.Decoder()
+    const buff = Buffer.from(decoder.write(key.toString().slice(1)).finalize())
+    return new CID(buff)
+  } catch (err) {
+    return { err: `Could not convert block with key '${key}' to CID: ${err.message}` }
+  }
+}
 
 exports.normalizePath = normalizePath
 exports.parseIpfsPath = parseIpfsPath
 exports.resolvePath = resolvePath
+exports.blockKeyToCid = blockKeyToCid

src/core/utils.js

@@ -32,7 +32,7 @@ function createNode (callback) {
   }, callback)
 }
 
-describe('kad-dht is routing content and peers correctly', () => {
+describe.only('kad-dht is routing content and peers correctly', () => {
   let nodeA
   let nodeB
   let nodeC