Merge branch 'master' into fix-client-node-tests

Author: ryanio

package-lock.json

@@ -736,15 +736,16 @@ export default class Blockchain implements BlockchainInterface {
    *
    * @param name - Name of the state root head
    * @param onBlock - Function called on each block with params (block, reorg)
+   * @param maxBlocks - How many blocks to run. By default, run all unprocessed blocks in the canonical chain.
    */
-  async iterator(name: string, onBlock: OnBlock) {
-    return this._iterator(name, onBlock)
+  async iterator(name: string, onBlock: OnBlock, maxBlocks?: number) {
+    return this._iterator(name, onBlock, maxBlocks)
   }
 
   /**
    * @hidden
    */
-  private async _iterator(name: string, onBlock: OnBlock) {
+  private async _iterator(name: string, onBlock: OnBlock, maxBlocks?: number) {
     await this.initAndLock<void>(async () => {
       const blockHash = this._heads[name] || this._genesis
       let lastBlock: Block | undefined
@@ -753,11 +754,15 @@ export default class Blockchain implements BlockchainInterface {
         return
       }
 
+      if (maxBlocks && maxBlocks < 0) {
+        throw 'If maxBlocks is provided, it has to be a non-negative number'
+      }
+
       const number = await this.dbManager.hashToNumber(blockHash)
       const blockNumber = number.addn(1)
+      let blocksRanCounter = 0
 
-      // eslint-disable-next-line no-constant-condition
-      while (true) {
+      while (maxBlocks !== blocksRanCounter) {
         try {
           const block = await this._getBlock(blockNumber)
 
@@ -770,6 +775,7 @@ export default class Blockchain implements BlockchainInterface {
           lastBlock = block
           await onBlock(block, reorg)
           blockNumber.iaddn(1)
+          blocksRanCounter++
         } catch (error) {
           if (error.type !== 'NotFoundError') {
             throw error
@@ -782,6 +788,19 @@ export default class Blockchain implements BlockchainInterface {
     })
   }
 
+  /**
+   * Set header hash of a certain `tag`.
+   * When calling the iterator, the iterator will start running the first child block after the header hash currenntly stored.
+   * @param tag - The tag to save the headHash to
+   * @param headHash - The head hash to save
+   */
+  async setHead(tag: string, headHash: Buffer) {
+    await this.initAndLock<void>(async () => {
+      this._heads[tag] = headHash
+      await this._saveHeads()
+    })
+  }
+
   /* Methods regarding re-org operations */
 
   /**

packages/blockchain/src/index.ts

@@ -348,6 +348,99 @@ tape('blockchain test', (t) => {
     st.end()
   })
 
+  t.test(
+    'should iterate through maxBlocks blocks if maxBlocks parameter is provided',
+    async (st) => {
+      const { blockchain, blocks, error } = await generateBlockchain(25)
+      st.error(error, 'no error')
+      let i = 0
+      await blockchain.iterator(
+        'test',
+        (block: Block) => {
+          if (block.hash().equals(blocks[i + 1].hash())) {
+            i++
+          }
+        },
+        5
+      )
+      st.equals(i, 5)
+      st.end()
+    }
+  )
+
+  t.test(
+    'should iterate through 0 blocks in case 0 maxBlocks parameter is provided',
+    async (st) => {
+      const { blockchain, blocks, error } = await generateBlockchain(25)
+      st.error(error, 'no error')
+      let i = 0
+      await blockchain
+        .iterator(
+          'test',
+          (block: Block) => {
+            if (block.hash().equals(blocks[i + 1].hash())) {
+              i++
+            }
+          },
+          0
+        )
+        .catch(() => {
+          st.fail('Promise cannot throw when running 0 blocks')
+        })
+      st.equals(i, 0)
+      st.end()
+    }
+  )
+
+  t.test('should throw on a negative maxBlocks parameter in iterator', async (st) => {
+    const { blockchain, blocks, error } = await generateBlockchain(25)
+    st.error(error, 'no error')
+    let i = 0
+    await blockchain
+      .iterator(
+        'test',
+        (block: Block) => {
+          if (block.hash().equals(blocks[i + 1].hash())) {
+            i++
+          }
+        },
+        -1
+      )
+      .catch(() => {
+        st.end()
+      })
+    // Note: if st.end() is not called (Promise did not throw), then this test fails, as it does not end.
+  })
+
+  t.test('should test setHead method', async (st) => {
+    const { blockchain, blocks, error } = await generateBlockchain(25)
+    st.error(error, 'no error')
+
+    const headBlockIndex = 5
+
+    const headHash = blocks[headBlockIndex].hash()
+    await blockchain.setHead('myHead', headHash)
+    const currentHeadBlock = await blockchain.getHead('myHead')
+
+    st.ok(headHash.equals(currentHeadBlock.hash()), 'head hash equals the provided head hash')
+
+    let i = 0
+    // check that iterator starts from this head block
+    await blockchain.iterator(
+      'myHead',
+      (block: Block) => {
+        if (block.hash().equals(blocks[headBlockIndex + 1].hash())) {
+          i++
+        }
+      },
+      5
+    )
+
+    st.equals(i, 1)
+
+    st.end()
+  })
+
   t.test('should catch iterator func error', async (st) => {
     const { blockchain, error } = await generateBlockchain(25)
     st.error(error, 'no error')

packages/blockchain/test/index.ts

@@ -5,6 +5,8 @@ import { parseMultiaddrs } from '../../util'
 import { Libp2pSender } from '../protocol/libp2psender'
 import { Peer, PeerOptions } from './peer'
 import { Libp2pNode } from './libp2pnode'
+import { Protocol } from '../protocol'
+import { Libp2pServer } from '../server'
 
 export interface Libp2pPeerOptions extends Omit<PeerOptions, 'address' | 'transport'> {
   /* Multiaddrs to listen on (can be a comma separated string or list) */
@@ -73,7 +75,7 @@ export class Libp2pPeer extends Peer {
    * @private
    * @return {Promise}
    */
-  async accept(protocol: any, connection: any, server: any): Promise<void> {
+  async accept(protocol: Protocol, connection: any, server: Libp2pServer): Promise<void> {
     await this.bindProtocol(protocol, new Libp2pSender(connection))
     this.inbound = true
     this.server = server
@@ -87,7 +89,7 @@ export class Libp2pPeer extends Peer {
    * @param  {Server}     [server] optional server that initiated connection
    * @return {Promise}
    */
-  async bindProtocols(node: any, peerInfo: any, server: any = null): Promise<void> {
+  async bindProtocols(node: Libp2pNode, peerInfo: any, server?: Libp2pServer): Promise<void> {
     await Promise.all(
       this.protocols.map(async (p: any) => {
         await p.open()

packages/client/lib/net/peer/libp2ppeer.ts

@@ -99,15 +99,14 @@ export class Peer extends events.EventEmitter {
    * @param  {Sender}    sender Sender instance provided by subclass
    * @return {Promise}
    * @example
-   *
    * await peer.bindProtocol(ethProtocol, sender)
    * // Example: Directly call message name as a method on the bound protocol
-   * const headers1 = await peer.eth.getBlockHeaders(1, 100, 0, 0)
+   * const headers1 = await peer.eth.getBlockHeaders({ block: new BN(1), max: 100 })
    * // Example: Call request() method with message name as first parameter
-   * const headers2 = await peer.eth.request('getBlockHeaders', 1, 100, 0, 0)
+   * const headers2 = await peer.eth.request('getBlockHeaders', { block: new BN(1), max: 100 })
    * // Example: Call send() method with message name as first parameter and
    * // wait for response message as an event
-   * peer.eth.send('getBlockHeaders', 1, 100, 0, 0)
+   * peer.eth.send('getBlockHeaders', { block: new BN(1), max: 100 })
    * peer.eth.on('message', ({ data }) => console.log(`Received ${data.length} headers`))
    */
   async bindProtocol(protocol: Protocol, sender: Sender): Promise<void> {

packages/client/lib/net/peer/peer.ts

@@ -1,18 +1,18 @@
 import { randomBytes } from 'crypto'
 import {
   Capabilities as Devp2pCapabilities,
-  DPT as Devp2pDPT,
   ETH as Devp2pETH,
   LES as Devp2pLES,
   Peer as Devp2pRlpxPeer,
   RLPx as Devp2pRLPx,
-} from 'ethereumjs-devp2p'
+} from '@ethereumjs/devp2p'
 import { Protocol, RlpxSender } from '../protocol'
 import { Peer, PeerOptions } from './peer'
+import { RlpxServer } from '../server'
 
 const devp2pCapabilities: any = {
-  eth62: Devp2pETH.eth62,
   eth63: Devp2pETH.eth63,
+  eth64: Devp2pETH.eth64,
   les2: Devp2pLES.les2,
 }
 
@@ -103,9 +103,7 @@ export class RlpxPeer extends Peer {
     await Promise.all(this.protocols.map((p) => p.open()))
     this.rlpx = new Devp2pRLPx(key, {
       capabilities: RlpxPeer.capabilities(this.protocols),
-      listenPort: null,
-      dpt: (<unknown>null) as Devp2pDPT, // TODO: required option
-      maxPeers: (<unknown>null) as number, // TODO: required option
+      common: this.config.common,
     })
     await this.rlpx.connect({
       id: Buffer.from(this.id, 'hex'),
@@ -143,7 +141,7 @@ export class RlpxPeer extends Peer {
    * @private
    * @return {Promise}
    */
-  async accept(rlpxPeer: Devp2pRlpxPeer, server: any): Promise<void> {
+  async accept(rlpxPeer: Devp2pRlpxPeer, server: RlpxServer): Promise<void> {
     if (this.connected) {
       return
     }
@@ -160,9 +158,9 @@ export class RlpxPeer extends Peer {
   async bindProtocols(rlpxPeer: Devp2pRlpxPeer): Promise<void> {
     this.rlpxPeer = rlpxPeer
     await Promise.all(
-      rlpxPeer.getProtocols().map((rlpxProtocol: any) => {
+      rlpxPeer.getProtocols().map((rlpxProtocol) => {
         const name = rlpxProtocol.constructor.name.toLowerCase()
-        const protocol = this.protocols.find((p: any) => p.name === name)
+        const protocol = this.protocols.find((p) => p.name === name)
         if (protocol) {
           return this.bindProtocol(protocol, new RlpxSender(rlpxProtocol))
         }

packages/client/lib/net/peer/rlpxpeer.ts

@@ -1,5 +1,5 @@
 import { EventEmitter } from 'events'
-import { Protocol } from '../protocol/protocol'
+import { Message, Protocol } from '../protocol/protocol'
 import { Peer } from '../peer/peer'
 import { Sender } from './sender'
 import { Config } from '../../config'
@@ -31,7 +31,7 @@ export class BoundProtocol extends EventEmitter {
   private versions: number[]
   private timeout: number
   private _status: any
-  private resolvers: Map<string, any>
+  private resolvers: Map<string | number, any>
 
   /**
    * Create bound protocol
@@ -80,9 +80,9 @@ export class BoundProtocol extends EventEmitter {
    * @param  {Object} message message object
    * @emits  message
    */
-  handle(incoming: any) {
+  handle(incoming: Message) {
     const messages = this.protocol.messages
-    const message = messages.find((m: any) => m.code === incoming.code)
+    const message = messages.find((m) => m.code === incoming.code)
     if (!message) {
       return
     }
@@ -117,16 +117,16 @@ export class BoundProtocol extends EventEmitter {
    * @param  name message name
    * @param  args message arguments
    */
-  send(name: string, args?: any): any {
+  send(name: string, args?: any) {
     const messages = this.protocol.messages
-    const message = messages.find((m: any) => m.name === name)
+    const message = messages.find((m) => m.name === name)
     if (message) {
       const encoded = this.protocol.encode(message, args)
       this.sender.sendMessage(message.code, encoded)
-      return message
     } else {
       throw new Error(`Unknown message: ${name}`)
     }
+    return message
   }
 
   /**
@@ -143,14 +143,14 @@ export class BoundProtocol extends EventEmitter {
       resolve: null,
       reject: null,
     }
-    if (this.resolvers.get(message.response)) {
+    if (this.resolvers.get(message.response!)) {
       throw new Error(`Only one active request allowed per message type (${name})`)
     }
-    this.resolvers.set(message.response, resolver)
+    this.resolvers.set(message.response!, resolver)
     return new Promise((resolve, reject) => {
       resolver.timeout = setTimeout(() => {
         resolver.timeout = null
-        this.resolvers.delete(message.response)
+        this.resolvers.delete(message.response!)
         reject(new Error(`Request timed out after ${this.timeout}ms`))
       }, this.timeout)
       resolver.resolve = resolve
@@ -163,13 +163,14 @@ export class BoundProtocol extends EventEmitter {
    * corresponding response message.
    */
   addMethods() {
-    const messages = this.protocol.messages.filter((m: any) => m.response)
+    const messages = this.protocol.messages.filter((m) => m.response)
     for (const message of messages) {
-      const name = message.name as string
+      const name = message.name
       const camel = name[0].toLowerCase() + name.slice(1)
       ;(this as any)[camel] = async (args: any[]) =>
         this.request(name, args).catch((error: Error) => {
           this.emit('error', error)
+          return []
         })
     }
   }