devp2p -> connection reliability: distribute network traffic on DPT additions of new neighbour peers

Author: holgerd77

packages/devp2p/src/dpt/dpt.ts

@@ -77,7 +77,7 @@ export class DPT extends EventEmitter {
     })
     this._server.once('listening', () => this.emit('listening'))
     this._server.once('close', () => this.emit('close'))
-    this._server.on('peers', (peers, remote) => this._onServerPeers(peers))
+    this._server.on('peers', (peers) => this._onServerPeers(peers))
     this._server.on('error', (err) => this.emit('error', err))
 
     const refreshIntervalSubdivided = Math.floor((options.refreshInterval ?? ms('60s')) / 10)
@@ -115,12 +115,16 @@ export class DPT extends EventEmitter {
   }
 
   _onServerPeers(peers: PeerInfo[]): void {
-    const ms = 0
+    const DIFF_TIME_MS = 200
+    let ms = 0
     for (const peer of peers) {
-      this.addPeer(peer).catch((error) => {
-        this.emit('error', error  )
-      })
-    } 
+      setTimeout(() => {
+        this.addPeer(peer).catch((error) => {
+          this.emit('error', error)
+        })
+      }, ms)
+      ms += DIFF_TIME_MS
+    }
   }
 
   async bootstrap(peer: PeerInfo): Promise<void> {

packages/devp2p/src/dpt/kbucket.ts

@@ -32,7 +32,7 @@ export class KBucket extends EventEmitter {
       this.emit('removed', peer)
     })
 
-    this._kbucket.on('ping', (oldPeers: PeerInfo[], newPeer: PeerInfo | undefined) => {
+    this._kbucket.on('ping', (oldPeers: PeerInfo[], newPeer: PeerInfo | undefined) => {
       this.emit('ping', oldPeers, newPeer)
     })
   }

packages/devp2p/test/integration/dpt-simulator.ts

@@ -32,7 +32,7 @@ test('DPT: remove node', async (t) => {
   async.series(
     [
       function (cb) {
-        dpts[0].on('peer:added', function (peer: any) {
+        dpts[0].on('peer:added', function (peer) {
           dpts[0].removePeer(peer)
           cb(null)
         })
@@ -64,13 +64,13 @@ test('DPT: ban node', async (t) => {
   async.series(
     [
       function (cb) {
-        dpts[0].on('peer:added', function (peer: any) {
+        dpts[0].on('peer:added', function (peer) {
           dpts[0].banPeer(peer)
           cb(null)
         })
       },
       function (cb) {
-        dpts[0].on('peer:removed', function (peer: any) {
+        dpts[0].on('peer:removed', function (peer) {
           t.equal(dpts[0].banlist.has(peer), true, 'ban-list should contain peer')
           t.equal(
             dpts[0].getPeers().length,
@@ -150,11 +150,14 @@ test('DPT: simulate bootstrap', async (t) => {
   }
 
   await delay(250)
-  util.destroyDPTs(dpts)
 
   // dpts.forEach((dpt, i) => console.log(`${i}:${dpt.getPeers().length}`))
-  for (const dpt of dpts)
+  for (const dpt of dpts) {
     t.equal(dpt.getPeers().length, numDPTs, 'Peers should be distributed to all DPTs')
+  }
+  await delay(1000)
+
+  util.destroyDPTs(dpts)
 
   t.end()
 })

packages/devp2p/test/integration/util.ts

@@ -5,7 +5,7 @@ import Common from '@ethereumjs/common'
 export const localhost = '127.0.0.1'
 export const basePort = 30306
 
-export function getTestDPTs(numDPTs: any) {
+export function getTestDPTs(numDPTs: number) {
   const dpts = []
 
   for (let i = 0; i < numDPTs; ++i) {
@@ -30,7 +30,7 @@ export function initTwoPeerDPTSetup() {
   return dpts
 }
 
-export function destroyDPTs(dpts: any) {
+export function destroyDPTs(dpts: DPT[]) {
   for (const dpt of dpts) dpt.destroy()
 }