Merge pull request #133 from ethereumjs/improve-sync-reliability

Improve sync reliability

lib/net/peerpool.js

@@ -43,6 +43,7 @@ class PeerPool extends EventEmitter {
     this.logger = options.logger
     this.maxPeers = options.maxPeers
     this.pool = new Map()
+    this.noPeerPeriods = 0
     this.init()
   }
 
@@ -63,6 +64,7 @@ class PeerPool extends EventEmitter {
       server.on('disconnected', (peer) => { this.disconnected(peer) })
     })
     this.opened = true
+    this._statusCheckInterval = setInterval(await this._statusCheck.bind(this), 20000)
   }
 
   /**
@@ -72,6 +74,7 @@ class PeerPool extends EventEmitter {
   async close () {
     this.pool.clear()
     this.opened = false
+    clearInterval(this._statusCheckInterval)
   }
 
   /**
@@ -185,6 +188,29 @@ class PeerPool extends EventEmitter {
       }
     }
   }
+
+  /**
+   * Peer pool status check on a repeated interval
+   */
+  async _statusCheck () {
+    if (this.size === 0) {
+      this.noPeerPeriods += 1
+      if (this.noPeerPeriods >= 3) {
+        const promises = this.servers.map(async server => {
+          if (server.bootstrap) {
+            this.logger.info('Retriggering bootstrap.')
+            await server.bootstrap()
+          }
+        })
+        await Promise.all(promises)
+        this.noPeerPeriods = 0
+      } else {
+        this.logger.info('Looking for suited peers...')
+      }
+    } else {
+      this.noPeerPeriods = 0
+    }
+  }
 }
 
 module.exports = PeerPool

lib/net/protocol/ethprotocol.js

@@ -46,7 +46,7 @@ class EthProtocol extends Protocol {
    * Create eth protocol
    * @param {Object}   options constructor parameters
    * @param {Chain}    options.chain blockchain
-   * @param {number}   [options.timeout=5000] handshake timeout in ms
+   * @param {number}   [options.timeout=8000] handshake timeout in ms
    * @param {Logger}   [options.logger] logger instance
    */
   constructor (options) {

lib/net/protocol/lesprotocol.js

@@ -65,7 +65,7 @@ class LesProtocol extends Protocol {
    * @param {Chain}       options.chain blockchain
    * @param {FlowControl} [options.flow] flow control manager. if undefined,
    * header serving will be disabled
-   * @param {number}      [options.timeout=5000] handshake timeout in ms
+   * @param {number}      [options.timeout=8000] handshake timeout in ms
    * @param {Logger}      [options.logger] logger instance
    */
   constructor (options) {

lib/net/protocol/protocol.js

@@ -6,7 +6,7 @@ const { defaultLogger } = require('../../logging')
 
 const defaultOptions = {
   logger: defaultLogger,
-  timeout: 5000
+  timeout: 8000
 }
 
 /**
@@ -28,7 +28,7 @@ class Protocol extends EventEmitter {
   /**
    * Create new protocol
    * @param {Object}   options constructor parameters
-   * @param {number}   [options.timeout=5000] handshake timeout in ms
+   * @param {number}   [options.timeout=8000] handshake timeout in ms
    * @param {Logger}   [options.logger] logger instance
    */
   constructor (options) {

lib/net/server/rlpxserver.js

@@ -91,16 +91,25 @@ class RlpxServer extends Server {
     this.initDpt()
     this.initRlpx()
 
-    this.bootnodes.map(node => {
+    await this.bootstrap()
+
+    this.started = true
+  }
+
+  /**
+   * Bootstrap bootnode peers from the network
+   * @return {Promise}
+   */
+  async bootstrap () {
+    const promises = this.bootnodes.map(node => {
       const bootnode = {
         address: node.ip,
         udpPort: node.port,
         tcpPort: node.port
       }
       return this.dpt.bootstrap(bootnode).catch(e => this.error(e))
     })
-
-    this.started = true
+    await Promise.all(promises)
   }
 
   /**

lib/service/ethereumservice.js

@@ -9,7 +9,7 @@ const defaultOptions = {
   lightserv: false,
   common: new Common('mainnet', 'chainstart'),
   minPeers: 3,
-  timeout: 5000,
+  timeout: 8000,
   interval: 1000
 }
 

lib/service/service.js

@@ -109,10 +109,13 @@ class Service extends EventEmitter {
   }
 
   /**
-   * Start service
+   * Stop service
    * @return {Promise}
    */
   async stop () {
+    if (this.opened) {
+      await this.close()
+    }
     this.running = false
     this.logger.info(`Stopped ${this.name} service.`)
   }

lib/sync/fetcher/fetcher.js

@@ -8,7 +8,7 @@ const { defaultLogger } = require('../../logging')
 const defaultOptions = {
   common: new Common('mainnet', 'chainstart'),
   logger: defaultLogger,
-  timeout: 5000,
+  timeout: 8000,
   interval: 1000,
   banTime: 60000,
   maxQueue: 16,

test/blockchain/chain.js

@@ -43,7 +43,7 @@ tape('[Chain]', t => {
     t.equal(chain.genesis.hash.toString('hex'),
       chain.blocks.latest.hash().toString('hex'),
       'get chain.block.latest')
-    chain.close()
+    await chain.close()
     t.end()
   })
 
@@ -80,6 +80,7 @@ tape('[Chain]', t => {
     await chain.getTd(block.hash())
     t.ok(chain.opened, 'chain should open if getTd() called')
     t.equal(await chain.open(), false, 'skip open if already opened')
+    await chain.close()
     t.end()
   })
 
@@ -89,6 +90,7 @@ tape('[Chain]', t => {
     t.notOk(await chain.putBlocks(), 'add undefined block')
     t.notOk(await chain.putBlocks(null), 'add null block')
     t.notOk(await chain.putBlocks([]), 'add empty block list')
+    await chain.close()
     t.end()
   })
 
@@ -98,6 +100,7 @@ tape('[Chain]', t => {
     t.notOk(await chain.putHeaders(), 'add undefined header')
     t.notOk(await chain.putHeaders(null), 'add null header')
     t.notOk(await chain.putHeaders([]), 'add empty header list')
+    await chain.close()
     t.end()
   })
 
@@ -112,7 +115,7 @@ tape('[Chain]', t => {
     await chain.putBlocks([block])
     t.equal(chain.blocks.td.toString(16), '4abcdffff', 'get chain.td')
     t.equal(chain.blocks.height.toString(10), '1', 'get chain.height')
-    chain.close()
+    await chain.close()
     t.end()
   })
 })

test/integration/peerpool.js

@@ -18,6 +18,7 @@ tape('[Integration:PeerPool]', async (t) => {
 
   async function destroy (server, pool) {
     await server.stop()
+    await pool.close()
   }
 
   t.test('should open', async (t) => {

test/service/fastethereumservice.js

@@ -57,6 +57,7 @@ tape('[FastEthereumService]', t => {
     service.pool.emit('added', 'peer0')
     service.pool.emit('removed', 'peer0')
     service.pool.emit('error', 'error1')
+    await service.close()
   })
 
   t.test('should start/stop', async (t) => {
@@ -69,6 +70,7 @@ tape('[FastEthereumService]', t => {
     td.verify(service.synchronizer.stop())
     td.verify(server.start())
     t.notOk(await service.stop(), 'already stopped')
+    await server.stop()
     t.end()
   })