devp2p -> connection reliability: RLPx type improvements

Author: holgerd77

packages/devp2p/src/rlpx/peer.ts

@@ -529,6 +529,7 @@ export class Peer extends EventEmitter {
             break
         }
       } catch (err) {
+        this.disconnect(DISCONNECT_REASONS.SUBPROTOCOL_ERROR)
         debug(`Error on peer socket data handling: ${err}`)
         this.emit('error', err)
       }

packages/devp2p/src/rlpx/rlpx.ts

@@ -80,9 +80,11 @@ export class RLPx extends EventEmitter {
         if (this._getOpenSlots() > 0) return this._connectToPeer(peer)
         this._peersQueue.push({ peer: peer, ts: 0 }) // save to queue
       })
-      this._dpt.on('peer:removed', (peer: any) => {
+      this._dpt.on('peer:removed', (peer: PeerInfo) => {
         // remove from queue
-        this._peersQueue = this._peersQueue.filter((item: any) => !item.peer.id.equals(peer.id))
+        this._peersQueue = this._peersQueue.filter(
+          (item) => !(item.peer.id! as Buffer).equals(peer.id as Buffer)
+        )
       })
     }
 
@@ -259,12 +261,12 @@ export class RLPx extends EventEmitter {
   _refillConnections() {
     if (!this._isAlive()) return
     debug(
-      `refill connections.. queue size: ${this._peersQueue.length}, peers: ${
-        this._peers.size
+      `refill connections.. peers: ${this._peers.size}, queue size: ${
+        this._peersQueue.length
       }, open slots: ${this._getOpenSlots()}`
     )
 
-    this._peersQueue = this._peersQueue.filter((item: any) => {
+    this._peersQueue = this._peersQueue.filter((item) => {
       if (this._getOpenSlots() === 0) return true
       if (item.ts > Date.now()) return true
 

packages/devp2p/src/util.ts

@@ -3,6 +3,8 @@ import { randomBytes } from 'crypto'
 import { privateKeyVerify, publicKeyConvert } from 'secp256k1'
 import createKeccakHash from 'keccak'
 import { decode } from 'rlp'
+import { ETH } from './eth'
+import { LES } from './les'
 
 export function keccak256(...buffers: Buffer[]) {
   const buffer = Buffer.concat(buffers)
@@ -53,7 +55,14 @@ export function xor(a: Buffer, b: any): Buffer {
   return buffer
 }
 
-export function assertEq(expected: any, actual: any, msg: string, debug: any): void {
+type assertInput = Buffer | Buffer[] | ETH.StatusMsg | LES.Status | number | null
+
+export function assertEq(
+  expected: assertInput,
+  actual: assertInput,
+  msg: string,
+  debug: Function
+): void {
   let message
   if (Buffer.isBuffer(expected) && Buffer.isBuffer(actual)) {
     if (expected.equals(actual)) return