Persist DHT nodes

docs/api.md

@@ -53,12 +53,14 @@ If `opts` is specified, then the default options (shown below) will be overridde
 
 ```js
 {
-  maxConns: Number,        // Max number of connections per torrent (default=55)
-  nodeId: String|Buffer,   // DHT protocol node ID (default=randomly generated)
-  peerId: String|Buffer,   // Wire protocol peer ID (default=randomly generated)
-  tracker: Boolean|Object, // Enable trackers (default=true), or options object for Tracker
-  dht: Boolean|Object,     // Enable DHT (default=true), or options object for DHT
-  webSeeds: Boolean        // Enable BEP19 web seeds (default=true)
+  maxConns: Number,         // Max number of connections per torrent (default=55)
+  nodeId: String|Buffer,    // DHT protocol node ID (default=randomly generated)
+  peerId: String|Buffer,    // Wire protocol peer ID (default=randomly generated)
+  tracker: Boolean|Object,  // Enable trackers (default=true), or options object for Tracker
+  dht: Boolean|Object,      // Enable DHT (default=true), or options object for DHT
+  persistDht: Boolean,      // Persist DHT nodes (default=true)
+  persistDhtPath: String,   // DHT persist save file (default=dht.json under OS appdata dir)
+  webSeeds: Boolean         // Enable BEP19 web seeds (default=true)
 }
 ```
 

index.js

@@ -2,6 +2,7 @@
 
 module.exports = WebTorrent
 
+var appDataFolder = require('app-data-folder')
 var Buffer = require('safe-buffer').Buffer
 var concat = require('simple-concat')
 var createTorrent = require('create-torrent')
@@ -19,6 +20,7 @@ var randombytes = require('randombytes')
 var speedometer = require('speedometer')
 var zeroFill = require('zero-fill')
 
+var dhtPersist = require('./lib/dhtpersist') // browser exclude
 var TCPPool = require('./lib/tcp-pool') // browser exclude
 var Torrent = require('./lib/torrent')
 
@@ -78,6 +80,9 @@ function WebTorrent (opts) {
   }
   self.nodeIdBuffer = Buffer.from(self.nodeId, 'hex')
 
+  // Default DHT persistence flag
+  if (!('persistDht' in opts)) opts.persistDht = true
+
   self._debugId = self.peerId.toString('hex').substring(0, 7)
 
   self.destroyed = false
@@ -123,8 +128,38 @@ function WebTorrent (opts) {
   self._uploadSpeed = speedometer()
 
   if (opts.dht !== false && typeof DHT === 'function' /* browser exclude */) {
+    var dhtOpts = extend({ nodeId: self.nodeId }, opts.dht)
+
+    if (opts.persistDht) {
+      // Construct state save location
+      self.dhtSaveFile =
+        opts.persistDhtPath ||
+        path.join(appDataFolder('webtorrent'), 'dht.json')
+
+      if (!dhtOpts.bootstrap) {
+        // Load persisted state
+        var nodes = dhtPersist.loadNodes(self.dhtSaveFile)
+        if (nodes) {
+          var bootstrap = []
+          for (var node of nodes) {
+            var nodeString = node.host + ':' + node.port
+            bootstrap.push(nodeString)
+          }
+          dhtOpts.bootstrap = bootstrap
+        }
+      }
+    }
+
     // use a single DHT instance for all torrents, so the routing table can be reused
-    self.dht = new DHT(extend({ nodeId: self.nodeId }, opts.dht))
+    self.dht = new DHT(dhtOpts)
+
+    if (opts.persistDht) {
+      // Persist state periodically
+      var saveInterval = 15 * 60 * 1000 // 15 minutes
+      self.saveDhtStateTimer = setInterval(function saveDhtState () {
+        dhtPersist.save(self.dht, self.dhtSaveFile)
+      }, saveInterval)
+    }
 
     self.dht.once('error', function (err) {
       self._destroy(err)
@@ -400,6 +435,20 @@ WebTorrent.prototype.address = function () {
     : { address: '0.0.0.0', family: 'IPv4', port: 0 }
 }
 
+/**
+ * Persist DHT state to disk.
+ * No effect if DHT is not loaded.
+ */
+WebTorrent.prototype.saveDhtState = function (cb) {
+  if (
+    this.dht !== false &&
+    this.dhtSaveFile &&
+    typeof DHT === 'function' /* browser exclude */
+  ) {
+    dhtPersist.save(this.dht, this.dhtSaveFile, cb)
+  }
+}
+
 /**
  * Destroy the client, including all torrents and connections to peers.
  * @param  {function} cb
@@ -428,6 +477,7 @@ WebTorrent.prototype._destroy = function (err, cb) {
 
   if (self.dht) {
     tasks.push(function (cb) {
+      clearInterval(self.saveDhtStateTimer)
       self.dht.destroy(cb)
     })
   }

lib/dhtpersist.js

@@ -0,0 +1,78 @@
+var fs = require('fs')
+var mkdirp = require('mkdirp')
+var path = require('path')
+
+var savingDhtState = false
+function saveDhtState (dht, file, cb) {
+  if (savingDhtState) return
+  if (!dht) return // Quell after destroy
+  savingDhtState = true
+  var dhtState = dht.toJSON()
+  var dhtStateJson = JSON.stringify(dhtState)
+  mkdirp(
+    path.dirname(file),
+    function handleDhtSaveDirCreated (err) {
+      if (err) {
+        savingDhtState = false
+        if (cb) cb(err)
+        return
+      }
+      fs.writeFile(
+        file,
+        dhtStateJson,
+        function handleDhtStateWritten () {
+          savingDhtState = false
+          if (cb) cb(null)
+        }
+      )
+    }
+  )
+}
+
+function readDhtState (file) {
+  try {
+    return fs.readFileSync(file)
+  } catch (e) {
+    switch (e.code) {
+      case 'EACCES':
+      case 'EISDIR':
+      case 'ENOENT':
+      case 'EPERM':
+        return null
+      default:
+        throw e
+    }
+  }
+}
+
+function parseDhtState (dhtStateJson) {
+  try {
+    return JSON.parse(dhtStateJson)
+  } catch (e) {
+    if (e instanceof SyntaxError) return null
+    else throw e
+  }
+}
+
+function loadDhtState (file) {
+  var dhtStateJson = readDhtState(file)
+  if (!dhtStateJson) return null
+  var dhtState = parseDhtState(dhtStateJson)
+  if (!dhtState) return null
+  return dhtState
+}
+
+function loadDhtNodes (file) {
+  var dhtState = loadDhtState(file)
+  if (!dhtState) return null
+  if (!('nodes' in dhtState)) return null
+  var nodes = dhtState.nodes
+  if (!Array.isArray(nodes)) return null
+  if (nodes.length === 0) return null // Don't load an empty nodes list
+  return nodes
+}
+
+module.exports = {
+  save: saveDhtState,
+  loadNodes: loadDhtNodes
+}

package.json

@@ -8,6 +8,7 @@
     "url": "https://webtorrent.io"
   },
   "browser": {
+    "./lib/dhtpersist.js": false,
     "./lib/server.js": false,
     "./lib/tcp-pool.js": false,
     "bittorrent-dht/client": false,
@@ -27,6 +28,7 @@
   },
   "dependencies": {
     "addr-to-ip-port": "^1.4.2",
+    "app-data-folder": "^1.0.0",
     "bitfield": "^2.0.0",
     "bittorrent-dht": "^8.0.0",
     "bittorrent-protocol": "^2.1.5",
@@ -40,6 +42,7 @@
     "load-ip-set": "^1.2.7",
     "memory-chunk-store": "^1.2.0",
     "mime": "^2.2.0",
+    "mkdirp": "^0.5.1",
     "multistream": "^2.0.5",
     "package-json-versionify": "^1.0.2",
     "parse-numeric-range": "^0.0.2",
@@ -84,6 +87,7 @@
     "serve-static": "^1.11.1",
     "standard": "*",
     "tape": "^4.6.0",
+    "tmp": "0.0.33",
     "webtorrent-fixtures": "^1.5.0"
   },
   "engines": {

test/node/persist-dht-nodes.js

@@ -0,0 +1,80 @@
+var test = require('tape')
+var tmp = require('tmp')
+var fs = require('fs')
+var networkAddress = require('network-address')
+var DHT = require('bittorrent-dht/server')
+var WebTorrent = require('../../')
+
+var loopback = '127.0.0.1'
+var localAddress = networkAddress.ipv4()
+var port = 9999
+
+test('Save DHT state', function (t) {
+  t.plan(4)
+  var saveFile = tmp.tmpNameSync()
+  var dhtServer = new DHT({ bootstrap: false })
+  dhtServer.on('error', function (err) { t.fail(err) })
+  dhtServer.on('warning', function (err) { t.fail(err) })
+  dhtServer.listen(port, function handleServerListening () {
+    var client = new WebTorrent({
+      dht: { bootstrap: false, host: localAddress },
+      persistDhtPath: saveFile
+    })
+    client.on('error', function (err) { t.fail(err) })
+    client.on('warning', function (err) { t.fail(err) })
+    client.dht.addNode({ host: loopback, port: port })
+    client.dht.on('node', function handleNodeAdded () {
+      client.saveDhtState(function handleDhtStateSaved () {
+        var dhtStateJson = fs.readFileSync(saveFile)
+        var dhtState = JSON.parse(dhtStateJson)
+        var nodes = dhtState.nodes
+        var node = nodes[0]
+        t.equal(node.host, loopback)
+        t.equal(node.port, port)
+        client.destroy(function handleClientDestroyed (err) {
+          t.error(err, 'client destroyed')
+        })
+        dhtServer.destroy(function handleDhtServerDestroyed (err) {
+          t.error(err, 'dht server destroyed')
+        })
+      })
+    })
+  })
+})
+
+test('Load DHT state', function (t) {
+  t.plan(4)
+  var saveFile = tmp.tmpNameSync()
+  var node = {
+    host: loopback,
+    port: port
+  }
+  var nodes = [ node ]
+  var dhtState = { nodes: nodes, values: {} }
+  var dhtStateJson = JSON.stringify(dhtState)
+  fs.writeFileSync(saveFile, dhtStateJson)
+  var dhtServer = new DHT({ bootstrap: false })
+  dhtServer.on('error', function (err) { t.fail(err) })
+  dhtServer.on('warning', function (err) { t.fail(err) })
+  dhtServer.listen(port, function handleServerListening () {
+    var client = new WebTorrent({
+      dht: { host: localAddress },
+      persistDhtPath: saveFile
+    })
+    client.on('error', function (err) { t.fail(err) })
+    client.on('warning', function (err) { t.fail(err) })
+    client.dht.on('ready', function handleReady () {
+      var dhtState = client.dht.toJSON()
+      var nodes = dhtState.nodes
+      var node = nodes[0]
+      t.equal(node.host, loopback)
+      t.equal(node.port, port)
+      client.destroy(function handleClientDestroyed (err) {
+        t.error(err, 'client destroyed')
+      })
+      dhtServer.destroy(function handleDhtServerDestroyed (err) {
+        t.error(err, 'dht server destroyed')
+      })
+    })
+  })
+})