On-disk database + warm bootstrap

index.js

@@ -8,9 +8,18 @@ const SocketPool = require('./lib/socket-pool')
 const Persistent = require('./lib/persistent')
 const Router = require('./lib/router')
 const Cache = require('xache')
+const NamespacedDB = require('namespaced-native')
 const Server = require('./lib/server')
 const connect = require('./lib/connect')
-const { FIREWALL, BOOTSTRAP_NODES, COMMANDS } = require('./lib/constants')
+const Snapshotter = require('./lib/snapshotter')
+const {
+  FIREWALL,
+  BOOTSTRAP_NODES,
+  COMMANDS,
+  DB_PATH,
+  DB_NS,
+  SNAPSHOT_KEYS
+} = require('./lib/constants')
 const { hash, createKeyPair } = require('./lib/crypto')
 const RawStreamSet = require('./lib/raw-stream-set')
 const ConnectionPool = require('./lib/connection-pool')
@@ -19,7 +28,8 @@ const { STREAM_NOT_CONNECTED } = require('./lib/errors')
 const DEFAULTS = {
   ...DHT.DEFAULTS,
   connectionKeepAlive: 5000,
-  randomPunchInterval: 20000
+  randomPunchInterval: 20000,
+  routingTableSnapshotInterval: 120000
 }
 
 class HyperDHT extends DHT {
@@ -47,6 +57,7 @@ class HyperDHT extends DHT {
     }
     this.rawStreams = new RawStreamSet(this)
     this.plugins = new Map()
+    this.db = new NamespacedDB({ path: opts.dbPath || DB_PATH })
 
     this._router = new Router(this, router)
     this._socketPool = new SocketPool(this, opts.host || '0.0.0.0')
@@ -61,6 +72,12 @@ class HyperDHT extends DHT {
     this._randomPunches = 0
     this._randomPunchLimit = 1 // set to one for extra safety for now
 
+    this._routingTableSnapshotter = null
+
+    this._warmBootstrapped = new Promise((resolve) => {
+      this._warmBootstrapDone = resolve
+    })
+
     this.once('persistent', () => {
       this._persistent = new Persistent(this, persistent)
       for (const plugin of this.plugins.values()) plugin.onpersistent()
@@ -74,6 +91,33 @@ class HyperDHT extends DHT {
       if (!this.online) return
       for (const server of this.listening) server.notifyOnline()
     })
+
+    this.once('ready', async () => {
+      if (this.destroyed) return
+      const k = b4a.from(SNAPSHOT_KEYS.ROUTING_TABLE)
+
+      try {
+        if (opts.warmBootstrap) await this._warmBootstrap(k)
+      } catch {
+        // Do nothing, warm bootstrap is best effort
+      } finally {
+        this._warmBootstrapDone()
+      }
+
+      if (this.destroyed) return
+
+      this._routingTableSnapshotter = new Snapshotter(
+        this,
+        DB_NS.SNAPSHOTS,
+        k,
+        opts.routingTableSnapshotInterval || DEFAULTS.routingTableSnapshotInterval,
+        () => {
+          return b4a.from(JSON.stringify(this.toArray({ limit: this.table.k })))
+        }
+      )
+
+      this._routingTableSnapshotter.start()
+    })
   }
 
   static DEFAULTS = DEFAULTS
@@ -94,18 +138,37 @@ class HyperDHT extends DHT {
     return new ConnectionPool(this)
   }
 
+  async _warmBootstrap(key) {
+    const n = await this.db.namespace(DB_NS.SNAPSHOTS)
+    const s = await n.get(key)
+    const cached = s !== null ? JSON.parse(b4a.toString(s)) : []
+
+    if (cached.length > 0) {
+      for (const node of cached) this.addNode(node)
+      await this.findNode(this.table.id).finished()
+    }
+  }
+
+  async fullyBootstrapped() {
+    await super.fullyBootstrapped()
+    await this.db.ready()
+    await this._warmBootstrapped
+  }
+
   async resume({ log = noop } = {}) {
     if (this._deferRandomPunch) this._lastRandomPunch = Date.now()
     await super.resume({ log })
     const resuming = []
     for (const server of this.listening) resuming.push(server.resume())
     log('Resuming hyperdht servers')
     await Promise.allSettled(resuming)
+    if (this._routingTableSnapshotter) this._routingTableSnapshotter.resume()
     log('Done, hyperdht fully resumed')
   }
 
   async suspend({ log = noop } = {}) {
     this._connectable = false // just so nothing gets connected during suspension
+    if (this._routingTableSnapshotter) this._routingTableSnapshotter.suspend()
     const suspending = []
     for (const server of this.listening) suspending.push(server.suspend())
     log('Suspending all hyperdht servers')
@@ -121,14 +184,17 @@ class HyperDHT extends DHT {
   }
 
   async destroy({ force = false } = {}) {
+    this._warmBootstrapDone()
     if (!force) {
       const closing = []
       for (const server of this.listening) closing.push(server.close())
       await Promise.allSettled(closing)
     }
     this._router.destroy()
     if (this._persistent) this._persistent.destroy()
-    for (const plugin of this.plugins.values()) plugin.destroy()
+    for (const plugin of this.plugins.values()) await plugin.destroy()
+    if (this._routingTableSnapshotter) await this._routingTableSnapshotter.stop()
+    await this.db.close({ force })
     await this.rawStreams.clear()
     await this._socketPool.destroy()
     await super.destroy()
@@ -517,9 +583,9 @@ class HyperDHT extends DHT {
     )
   }
 
-  register(name, plugin) {
+  async register(name, plugin) {
     this.plugins.set(name, plugin)
-    plugin.onregister(this)
+    await plugin.onregister(this)
   }
 }
 

lib/constants.js

@@ -62,3 +62,13 @@ exports.NS = {
   PEER_HOLEPUNCH: NS_PEER_HOLEPUNCH,
   PLUGIN: NS_PLUGIN
 }
+
+exports.DB_PATH = './hyperdht.db'
+
+exports.DB_NS = {
+  SNAPSHOTS: 'hyperdht-snapshots'
+}
+
+exports.SNAPSHOT_KEYS = {
+  ROUTING_TABLE: 'routing-table'
+}

lib/plugin.js

@@ -9,7 +9,7 @@ module.exports = class Plugin {
     this.dht = null
   }
 
-  onregister(dht) {
+  async onregister(dht) {
     this.dht = dht
   }
 
@@ -21,7 +21,7 @@ module.exports = class Plugin {
     throw new Error('onpersistent() must be implemented')
   }
 
-  destroy() {
+  async destroy() {
     throw new Error('destroy() must be implemented')
   }
 

lib/snapshotter.js

@@ -0,0 +1,60 @@
+const safetyCatch = require('safety-catch')
+const Signal = require('signal-promise')
+const Sleeper = require('./sleeper')
+
+module.exports = class Snapshotter {
+  constructor(dht, namespace, key, interval, snapshot) {
+    this.dht = dht
+    this.namespace = namespace
+    this.key = key
+    this.interval = interval
+    this.snapshot = snapshot
+    this.sleeper = new Sleeper()
+    this.stopped = false
+    this.suspended = false
+    this._resumed = new Signal()
+    this.done = new Promise((resolve) => {
+      this._onDone = resolve
+    })
+  }
+
+  async start() {
+    while (!this.dht.destroyed && !this.stopped) {
+      try {
+        const n = await this.dht.db.namespace(this.namespace)
+        if (!this.suspended) await n.put([{ key: this.key, value: await this.snapshot() }])
+      } catch (err) {
+        safetyCatch(err)
+      }
+
+      if (this.dht.destroyed || this.stopped) break
+
+      if (this.suspended) {
+        await this._resumed.wait()
+      } else {
+        await this.sleeper.pause(this.interval)
+      }
+    }
+
+    this._onDone()
+  }
+
+  suspend() {
+    if (this.suspended) return
+    this.suspended = true
+    this.sleeper.resume()
+  }
+
+  resume() {
+    if (!this.suspended) return
+    this.suspended = false
+    this._resumed.notify()
+  }
+
+  async stop() {
+    this.stopped = true
+    this.sleeper.resume()
+    this._resumed.notify()
+    await this.done
+  }
+}

test/all.js

@@ -15,10 +15,11 @@ async function runTests() {
   await import('./messages.js')
   await import('./nat.js')
   await import('./noncustodial.js')
+  await import('./plugins.js')
   await import('./pool.js')
   await import('./relaying.js')
+  await import('./snapshots.js')
   await import('./storing.js')
-  await import('./plugins.js')
 
   test.resume()
 }

test/announces.js

@@ -3,6 +3,7 @@ const b4a = require('b4a')
 const cenc = require('compact-encoding')
 const messages = require('../lib/messages.js')
 const { swarm, toArray } = require('./helpers')
+const tmp = require('test-tmp')
 const DHT = require('../')
 const HyperDHT = require('../')
 const { COMMANDS } = require('../lib/constants.js')
@@ -226,7 +227,7 @@ test('announcer background does not over-trigger', async function (t) {
   const testnet = await swarm(t, 2) // must be <=3 (less than announcer MIN_ACTIVE) to trigger previous bug
   const bootstrap = testnet.bootstrap
 
-  const a = new HyperDHT({ bootstrap })
+  const a = new HyperDHT({ bootstrap, dbPath: await tmp(t) })
 
   const initTid = a.io._tid
   const server = a.createServer()

test/connections.js

@@ -142,8 +142,8 @@ test('createServer + connect - force holepunch', async function (t) {
   const [boot] = await swarm(t)
 
   const bootstrap = [{ host: '127.0.0.1', port: boot.address().port }]
-  const a = createDHT({ bootstrap, quickFirewall: false, ephemeral: true })
-  const b = createDHT({ bootstrap, quickFirewall: false, ephemeral: true })
+  const a = await createDHT(t, { bootstrap, quickFirewall: false, ephemeral: true })
+  const b = await createDHT(t, { bootstrap, quickFirewall: false, ephemeral: true })
 
   await a.fullyBootstrapped()
   await b.fullyBootstrapped()
@@ -188,8 +188,8 @@ test('createServer + connect - failed LAN ping falls back to holepunch', async f
   // same-LAN shortcut. This does not model NAT; the injected ping failure
   // below verifies that a blocked LAN shortcut still falls through to normal
   // holepunch.
-  const a = createDHT({ bootstrap, quickFirewall: false, ephemeral: true })
-  const b = createDHT({ bootstrap, quickFirewall: false, ephemeral: true })
+  const a = await createDHT(t, { bootstrap, quickFirewall: false, ephemeral: true })
+  const b = await createDHT(t, { bootstrap, quickFirewall: false, ephemeral: true })
 
   await a.fullyBootstrapped()
   await b.fullyBootstrapped()
@@ -255,8 +255,8 @@ test('server choosing to abort holepunch', async function (t) {
   const [boot] = await swarm(t)
 
   const bootstrap = [{ host: '127.0.0.1', port: boot.address().port }]
-  const a = createDHT({ bootstrap, quickFirewall: false, ephemeral: true })
-  const b = createDHT({ bootstrap, quickFirewall: false, ephemeral: true })
+  const a = await createDHT(t, { bootstrap, quickFirewall: false, ephemeral: true })
+  const b = await createDHT(t, { bootstrap, quickFirewall: false, ephemeral: true })
 
   await a.fullyBootstrapped()
   await b.fullyBootstrapped()
@@ -303,8 +303,8 @@ test('client choosing to abort holepunch', async function (t) {
   const [boot] = await swarm(t)
 
   const bootstrap = [{ host: '127.0.0.1', port: boot.address().port }]
-  const a = createDHT({ bootstrap, quickFirewall: false, ephemeral: true })
-  const b = createDHT({ bootstrap, quickFirewall: false, ephemeral: true })
+  const a = await createDHT(t, { bootstrap, quickFirewall: false, ephemeral: true })
+  const b = await createDHT(t, { bootstrap, quickFirewall: false, ephemeral: true })
 
   await a.fullyBootstrapped()
   await b.fullyBootstrapped()
@@ -474,8 +474,8 @@ test('relayed connection', async function (t) {
 
   const nodes = await swarm(t)
 
-  const a = nodes.createNode()
-  const b = nodes.createNode()
+  const a = await nodes.createNode(t)
+  const b = await nodes.createNode(t)
 
   const server = a.createServer()
   await server.listen()
@@ -501,7 +501,7 @@ test('relayed connection on same node', async function (t) {
 
   const nodes = await swarm(t)
 
-  const a = nodes.createNode()
+  const a = await nodes.createNode(t)
 
   const server = a.createServer()
   await server.listen()
@@ -571,8 +571,8 @@ test('create many connections with reusable sockets', async function (t) {
   const [boot] = await swarm(t)
 
   const bootstrap = [{ host: '127.0.0.1', port: boot.address().port }]
-  const a = createDHT({ bootstrap, quickFirewall: false, ephemeral: true })
-  const b = createDHT({ bootstrap, quickFirewall: false, ephemeral: true })
+  const a = await createDHT(t, { bootstrap, quickFirewall: false, ephemeral: true })
+  const b = await createDHT(t, { bootstrap, quickFirewall: false, ephemeral: true })
 
   await a.fullyBootstrapped()
   await b.fullyBootstrapped()
@@ -794,8 +794,8 @@ test('connectionKeepAlive can be turned off', async function (t) {
 
   const { bootstrap } = await swarm(t)
 
-  const a = createDHT({ bootstrap, connectionKeepAlive: false })
-  const b = createDHT({ bootstrap, connectionKeepAlive: false })
+  const a = await createDHT(t, { bootstrap, connectionKeepAlive: false })
+  const b = await createDHT(t, { bootstrap, connectionKeepAlive: false })
   t.teardown(async () => {
     await a.destroy()
     await b.destroy()
@@ -823,8 +823,8 @@ test('connectionKeepAlive passed to server and connection', async function (t) {
 
   const { bootstrap } = await swarm(t)
 
-  const a = createDHT({ bootstrap, connectionKeepAlive: 10000 })
-  const b = createDHT({ bootstrap, connectionKeepAlive: 20000 })
+  const a = await createDHT(t, { bootstrap, connectionKeepAlive: 10000 })
+  const b = await createDHT(t, { bootstrap, connectionKeepAlive: 20000 })
 
   const server = a.createServer((socket) => {
     socket.on('end', () => socket.end())
@@ -850,7 +850,7 @@ test('connectionKeepAlive passed to server and connection', async function (t) {
 test('bootstrap with suggested-IP', async function (t) {
   const [boot] = await swarm(t, 1)
   const bootstrap = ['127.0.0.1@invalid:' + boot.address().port]
-  const a = createDHT({ bootstrap, quickFirewall: false, ephemeral: false })
+  const a = await createDHT(t, { bootstrap, quickFirewall: false, ephemeral: false })
   await a.fullyBootstrapped()
 
   t.alike(boot.toArray(), [{ host: '127.0.0.1', port: a.address().port }])
@@ -859,11 +859,11 @@ test('bootstrap with suggested-IP', async function (t) {
 })
 
 test('Populate DHT with options.nodes', async function (t) {
-  const a = createDHT({ bootstrap: [] })
+  const a = await createDHT(t, { bootstrap: [] })
   await a.fullyBootstrapped()
   const nodes = [{ host: '127.0.0.1', port: a.address().port }]
 
-  const b = createDHT({ nodes, bootstrap: [] })
+  const b = await createDHT(t, { nodes, bootstrap: [] })
   await b.fullyBootstrapped()
 
   t.alike(b.toArray(), [{ host: '127.0.0.1', port: a.address().port }])