Adaptive ephemerality

README.md

@@ -31,7 +31,8 @@ bind-to-address]
 Options:
   -b, --bind                 Listening host                             [string]
   --dp, --dht_port           DHT listening port              [number] [required]
-  --de, --dht_ephemeral      DHT node epemerality                       [number]
+  --de, --dht_ephemeral      DHT node epemerality                       [boolean]
+  --da, --dht_adaptive       DHT node adaptive ephemerality             [boolean]
   --dht_maxValues            DHT max values                             [number]
   --bn, --bootstrap          Bootstrap nodes                 [string] [required]
   --aph, --api_port          HTTP api port                   [number] [required]
@@ -74,6 +75,7 @@ g.start()
  - `options` <Object> Options for the link
     - `host` <String> IP to bind to. If null, Grape binds to all interfaces
     - `dht_ephemeral` <Boolean> Whether to join the DHT (false) or just query it (true). Default is false
+    - `dht_adaptive` <Boolean> Upgrade ephemeral node to persistent (join DHT) after 20-30 minutes uptime. Default is false
     - `dht_maxValues` <Number> Maximum number of DHT values
     - `dht_port` <Number> Port for DHT
     - `dht_bootstrap`: <Array> Bootstrap servers
@@ -96,12 +98,10 @@ Emitted when a potential peer is found.
 
 Emitted when the DHT finds a new node.
 
-
 #### Event: 'warning'
 
 Emitted when a warning occurs in the DHT.
 
-
 #### Event: 'announce'
 
 Emitted when a peer announces itself in order to be stored in the DHT.
@@ -110,6 +110,10 @@ Emitted when a peer announces itself in order to be stored in the DHT.
 
 Emitted when a peer unannounces itself in order to be removed from the DHT.
 
+#### Event: 'persistent'
+
+Emitted when a peer turns from ephemeral to persistent. Only applies when `dht_ephemeral` and `dht_adaptive` are both `true` at initialization.
+
 ## RPC API
 
 ### Immutable Get

bin/grape.js

@@ -19,7 +19,12 @@ const program = require('yargs')
   .option('de', {
     describe: 'DHT node epemerality',
     alias: 'dht_ephemeral',
-    type: 'number'
+    type: 'boolean'
+  })
+  .option('de', {
+    describe: 'DHT node adaptive ephemerality',
+    alias: 'dht_adaptive',
+    type: 'boolean'
   })
   .option('dht_maxValues', {
     describe: 'DHT max values',

lib/Grape.js

@@ -43,6 +43,7 @@ class Grape extends Events {
     this.conf = _.defaults({}, conf, {
       host: null,
       dht_ephemeral: false,
+      dht_adaptive: false,
       dht_maxValues: 5000,
       dht_port: 20001,
       dht_bootstrap: [],
@@ -51,6 +52,10 @@ class Grape extends Events {
       check_maxPayloadSize: 8192
     })
 
+    if (this.conf.dht_adaptive && this.conf.dht_ephemeral !== true) {
+      throw Error('dht_adaptive can only applied when dht_ephemeral: true')
+    }
+
     this._mem = records({
       maxSize: 5000,
       maxAge: Math.ceil(this.conf.dht_peer_maxAge / 2)
@@ -67,6 +72,8 @@ class Grape extends Events {
 
   createNode (cb) {
     const dht = this.dht = DHT({
+      ephemeral: this.conf.dht_ephemeral,
+      adaptive: this.conf.dht_adaptive,
       maxValues: this.conf.dht_maxValues,
       bootstrap: this.conf.dht_bootstrap,
       maxAge: this.conf.dht_peer_maxAge
@@ -106,6 +113,11 @@ class Grape extends Events {
       debug(this.conf.dht_port, 'error', err)
     })
 
+    dht.on('persistent', () => {
+      debug(this.conf.dht_port, 'this peer has become non-ephemeral')
+      this.emit('persistent')
+    })
+
     const handleInitError = (err) => {
       this._initError = new GrapeError('ERR_GRAPE_INIT: ' + err.message)
       cb(err, dht)
@@ -120,6 +132,10 @@ class Grape extends Events {
     })
   }
 
+  get ephemeral () {
+    return this.dht.ephemeral
+  }
+
   hex2str (val) {
     return Buffer.from(val, 'hex').toString()
   }

package.json

@@ -10,7 +10,7 @@
     "micro-services"
   ],
   "dependencies": {
-    "@hyperswarm/dht": "^3.1.0",
+    "@hyperswarm/dht": "git+https://github.com/hyperswarm/dht.git#adaptive-ephemerality",
     "@hyperswarm/hypersign": "^2.1.0",
     "debug": "^4.0.1",
     "lodash": "^4.17.11",
@@ -19,7 +19,7 @@
     "yargs": "^12.0.2"
   },
   "engine": {
-    "node": ">=8.0"
+    "node": ">=10.0"
   },
   "bin": {
     "grape": "bin/grape.js"

test/Grape.js

@@ -1,12 +1,12 @@
 'use strict'
 const Events = require('events')
-const { promisifyOf, when, once } = require('nonsynchronous')
+const { promisifyOf, when, timeout, once } = require('nonsynchronous')
 const { test, teardown } = require('tap')
 const getPort = require('get-port')
 const { Grape } = require('..')
 const stop = promisifyOf('stop')
 const start = promisifyOf('start')
-
+const { createBootstrap } = require('./helper.js')
 const guard = (grape) => teardown(() => grape.stop())
 
 test('Grape', async () => {
@@ -92,6 +92,26 @@ test('Grape', async () => {
     await until.done()
   })
 
+  test('dht_adaptive true when dht_ephemeral false', async ({ rejects, resolves }) => {
+    rejects(async () => {
+      new Grape({ // eslint-disable-line no-new
+        dht_port: await getPort(),
+        dht_ephemeral: false,
+        dht_adaptive: true
+      })
+    }, Error('dht_adaptive can only applied when dht_ephemeral: true'))
+    rejects(async () => {
+      new Grape({ // eslint-disable-line no-new
+        dht_port: await getPort(),
+        dht_adaptive: true
+      })
+    }, Error('dht_adaptive can only applied when dht_ephemeral: true'))
+    resolves(async () => {
+      const grape = new Grape({ dht_port: await getPort(), dht_adaptive: true, dht_ephemeral: true })
+      grape.stop()
+    })
+  })
+
   test('dht port collision', async ({ ok, is }) => {
     const grape1 = new Grape({
       dht_port: await getPort(),
@@ -231,3 +251,66 @@ test('Grape', async () => {
     await stop(grape)()
   })
 })
+
+test('adaptive ephemerality', async ({ is, pass, resolves, rejects, tearDown }) => {
+  const { setTimeout } = global
+  const { random } = Math
+  const divideTimeBy = 10000
+  const wait = (1000 * 60 * 25) / divideTimeBy
+  global.setTimeout = (fn, t, ...args) => {
+    return setTimeout(fn, t / divideTimeBy, ...args)
+  }
+  Math.random = () => 0.5
+  tearDown(() => {
+    global.setTimeout = setTimeout
+    Math.random = random
+    peer.stop()
+    adapt.stop()
+    bs.stop()
+  })
+
+  const bs = await createBootstrap(true)
+
+  const adapt = new Grape({
+    api_port: await getPort(),
+    dht_port: await getPort(),
+    dht_ephemeral: true,
+    dht_adaptive: true,
+    dht_bootstrap: bs.dht_bootstrap
+  })
+  adapt.start()
+  await once(adapt, 'ready')
+  const peer = new Grape({
+    api_port: await getPort(),
+    dht_port: await getPort(),
+    dht_ephemeral: true,
+    dht_bootstrap: bs.dht_bootstrap
+  })
+  peer.start()
+  const announce = promisifyOf('announce')
+  const lookup = promisifyOf('lookup')
+  await once(peer, 'ready')
+  const topic = 'rest:util:net'
+  await rejects(announce(peer)(topic, 1234), Error('No close nodes responded'), 'expected no nodes found')
+
+  const { setEphemeral } = adapt.dht
+  let setEphemeralCalled = false
+  adapt.dht.setEphemeral = (bool, cb) => {
+    setEphemeralCalled = true
+    is(bool, false)
+    return setEphemeral.call(adapt.dht, bool, cb)
+  }
+  is(adapt.ephemeral, true)
+  const dhtJoined = once(adapt, 'persistent')
+  resolves(dhtJoined, 'dht joined event fired')
+  is(setEphemeralCalled, false)
+  await timeout(wait)
+  is(setEphemeralCalled, true)
+  await dhtJoined
+  is(adapt.ephemeral, false)
+  await announce(peer)(topic, 1234)
+
+  lookup(peer)(topic)
+  const [, { referrer }] = await once(peer, 'peer')
+  is(Buffer.compare(referrer.id, adapt.dht.id), 0)
+})

test/helper.js

@@ -5,17 +5,26 @@ const getPort = require('get-port')
 const { teardown } = require('tap')
 exports.createGrapes = createGrapes
 
-async function createGrapes (n, onstart = () => {}) {
-  const grapes = []
+async function createBootstrap (ephemeral = false) {
   const bsPort = await getPort()
-  const bootstrap = new Grape({
+  const bs = new Grape({
+    dht_ephemeral: ephemeral,
     dht_port: bsPort,
     dht_bootstrap: false,
     api_port: await getPort(bsPort + 1),
     dht_peer_maxAge: 200
   })
-  bootstrap.start()
-  await once(bootstrap, 'ready')
+  bs.start()
+  await once(bs, 'ready')
+  bs.dht_bootstrap = [`127.0.0.1:${bsPort}`]
+  return bs
+}
+
+exports.createBootstrap = createBootstrap
+
+async function createGrapes (n, onstart = () => {}) {
+  const grapes = []
+  const bootstrap = await createBootstrap()
 
   const ports = async (n, offset = 0) => {
     const result = new Array(n)
@@ -31,7 +40,7 @@ async function createGrapes (n, onstart = () => {}) {
   for (const [i, port] of dhtPorts.entries()) {
     const grape = new Grape({
       dht_port: port,
-      dht_bootstrap: [`127.0.0.1:${bsPort}`],
+      dht_bootstrap: bootstrap.dht_bootstrap,
       api_port: apiPorts[i],
       dht_peer_maxAge: 200
     })