Merge pull request #1075 from ethereumjs/fix-client-sync-bugs

Client: Fix Sync Bugs and Error Messages

packages/block/src/block.ts

@@ -201,15 +201,17 @@ export class Block {
    * and do a check on the root hash.
    */
   async validateTransactionsTrie(): Promise<boolean> {
+    let result
     if (this.transactions.length === 0) {
-      return this.header.transactionsTrie.equals(KECCAK256_RLP)
+      result = this.header.transactionsTrie.equals(KECCAK256_RLP)
+      return result
     }
 
     if (this.txTrie.root.equals(KECCAK256_RLP)) {
       await this.genTxTrie()
     }
-
-    return this.txTrie.root.equals(this.header.transactionsTrie)
+    result = this.txTrie.root.equals(this.header.transactionsTrie)
+    return result
   }
 
   /**

packages/client/lib/blockchain/chain.ts

@@ -109,7 +109,7 @@ export class Chain extends EventEmitter {
       new Blockchain({
         db: options.chainDB,
         common: this.config.chainCommon,
-        validateBlocks: false,
+        validateBlocks: true,
         validateConsensus: false,
       })
 

packages/client/lib/net/server/rlpxserver.ts

@@ -13,17 +13,28 @@ export interface RlpxServerOptions extends ServerOptions {
 
 const ignoredErrors = new RegExp(
   [
-    'EPIPE',
+    // Peer socket connection
     'ECONNRESET',
-    'ETIMEDOUT',
-    'NetworkId mismatch',
-    'Timeout error: ping',
+    'EPIPE', // (?)
+    'ETIMEDOUT', // (?)
+
+    // ETH status handling
     'Genesis block mismatch',
-    'Handshake timed out',
+    'NetworkId mismatch',
+    'Unknown fork hash',
+
+    // DPT message decoding
+    'Hash verification failed',
     'Invalid address buffer',
-    'Invalid MAC',
     'Invalid timestamp buffer',
-    'Hash verification failed',
+    'Invalid type',
+    'Timeout error: ping', // connection
+
+    // ECIES message encryption
+    'Invalid MAC',
+
+    // Client
+    'Handshake timed out', // Protocol handshake
   ].join('|')
 )
 

packages/client/lib/sync/execution/execution.ts

@@ -46,6 +46,7 @@ export abstract class Execution extends EventEmitter {
    */
   async stop(): Promise<boolean> {
     this.running = false
+    this.config.logger.info('Stopped execution.')
     return true
   }
 }

packages/client/lib/sync/execution/vmexecution.ts

@@ -44,7 +44,7 @@ export class VMExecution extends Execution {
    * Initializes VM execution. Must be called before run() is called
    */
   async open(): Promise<void> {
-    const headBlock = await this.vm.blockchain.getHead()
+    const headBlock = await this.vm.blockchain.getIteratorHead()
     const blockNumber = headBlock.header.number.toNumber()
     this.config.execCommon.setHardforkByBlockNumber(blockNumber)
     this.hardfork = this.config.execCommon.hardfork()
@@ -71,6 +71,7 @@ export class VMExecution extends Execution {
 
     let headBlock: Block | undefined
     let parentState: Buffer | undefined
+    let errorBlock: Block | undefined
     while (
       (numExecuted === undefined || numExecuted === this.NUM_BLOCKS_PER_ITERATION) &&
       !startHeadBlock.hash().equals(canonicalHead.hash()) &&
@@ -79,10 +80,14 @@ export class VMExecution extends Execution {
     ) {
       headBlock = undefined
       parentState = undefined
+      errorBlock = undefined
 
       this.vmPromise = blockchain.iterator(
         'vm',
         async (block: Block, reorg: boolean) => {
+          if (errorBlock) {
+            return
+          }
           // determine starting state for block run
           // if we are just starting or if a chain re-org has happened
           if (!headBlock || reorg) {
@@ -112,10 +117,26 @@ export class VMExecution extends Execution {
             // set as new head block
             headBlock = block
           } catch (error) {
+            // TODO: determine if there is a way to differentiate between the cases
+            // a) a bad block is served by a bad peer -> delete the block and restart sync
+            //    sync from parent block
+            // b) there is a consensus error in the VM -> stop execution until the
+            //    consensus error is fixed
+            //
+            // For now only option b) is implemented, atm this is a very likely case
+            // and the implemented behavior helps on debugging.
+            // Option a) would likely need some block comparison of the same blocks
+            // received by different peer to decide on bad blocks
+            // (minimal solution: receive block from 3 peers and take block if there is
+            // is equally served from at least 2 peers)
+            /*try {
             // remove invalid block
-            await blockchain!.delBlock(block.header.hash())
-            const blockNumber = block.header.number.toNumber()
-            const hash = short(block.hash())
+              await blockchain!.delBlock(block.header.hash())
+            } catch (error) {
+              this.config.logger.error(
+                `Error deleting block number=${blockNumber} hash=${hash} on failed execution`
+              )
+            }
             this.config.logger.warn(
               `Deleted block number=${blockNumber} hash=${hash} on failed execution`
             )
@@ -126,14 +147,26 @@ export class VMExecution extends Execution {
                 `Set back hardfork along block deletion number=${blockNumber} hash=${hash} old=${this.hardfork} new=${hardfork}`
               )
               this.config.execCommon.setHardforkByBlockNumber(blockNumber)
-            }
-            throw error
+            }*/
+            // Option a): set iterator head to the parent block so that an
+            // error can repeatedly processed for debugging
+            const blockNumber = block.header.number.toNumber()
+            const hash = short(block.hash())
+            this.config.logger.warn(`Execution of block number=${blockNumber} hash=${hash} failed`)
+            this.emit('error', error)
+            errorBlock = block
           }
         },
         this.NUM_BLOCKS_PER_ITERATION
       )
       numExecuted = (await this.vmPromise) as number
 
+      // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition
+      if (errorBlock) {
+        await this.chain.blockchain.setIteratorHead('vm', (errorBlock as Block).header.parentHash)
+        return 0
+      }
+
       const endHeadBlock = await this.vm.blockchain.getHead()
       if (numExecuted > 0) {
         const firstNumber = startHeadBlock.header.number.toNumber()

packages/client/lib/sync/fullsync.ts

@@ -27,6 +27,11 @@ export class FullSynchronizer extends Synchronizer {
     })
 
     const self = this
+    this.execution.on('error', async (error: Error) => {
+      self.emit('error', error)
+      await self.stop()
+    })
+
     this.chain.on('updated', async function () {
       // for some reason, if we use .on('updated', this.runBlocks)
       // it runs in the context of the Chain and not in the FullSync context..?

packages/client/lib/sync/sync.ts

@@ -119,6 +119,7 @@ export abstract class Synchronizer extends EventEmitter {
     }
     await new Promise((resolve) => setTimeout(resolve, this.interval))
     this.running = false
+    this.config.logger.info('Stopped synchronization.')
     return true
   }
 }

packages/client/test/blockchain/chain.spec.ts

@@ -8,6 +8,7 @@ import { Config } from '../../lib/config'
 // needed for karma-typescript bundling
 import * as util from 'util' // eslint-disable-line @typescript-eslint/no-unused-vars
 import { Buffer } from 'buffer' // eslint-disable-line @typescript-eslint/no-unused-vars
+import Blockchain from '@ethereumjs/blockchain'
 
 const config = new Config()
 
@@ -42,7 +43,11 @@ tape('[Chain]', (t) => {
   })
 
   t.test('should detect unopened chain', async (t) => {
-    const chain = new Chain({ config })
+    const blockchain = new Blockchain({
+      validateBlocks: false,
+      validateConsensus: false,
+    })
+    const chain = new Chain({ config, blockchain })
     const headerData: HeaderData = {
       number: new BN(1),
       difficulty: new BN('abcdffff', 16),
@@ -81,7 +86,12 @@ tape('[Chain]', (t) => {
   })
 
   t.test('should add block to chain', async (t) => {
-    const chain = new Chain({ config })
+    // TODO: add test cases with activated block validation
+    const blockchain = new Blockchain({
+      validateBlocks: false,
+      validateConsensus: false,
+    })
+    const chain = new Chain({ config, blockchain })
     await chain.open()
     const headerData: HeaderData = {
       number: new BN(1),

packages/client/test/integration/fullethereumservice.spec.ts

@@ -5,13 +5,18 @@ import { FullEthereumService } from '../../lib/service'
 import MockServer from './mocks/mockserver'
 import MockChain from './mocks/mockchain'
 import { destroy } from './util'
+import Blockchain from '@ethereumjs/blockchain'
 
 tape('[Integration:FullEthereumService]', async (t) => {
   async function setup(): Promise<[MockServer, FullEthereumService]> {
     const loglevel = 'error'
     const config = new Config({ loglevel })
     const server = new MockServer({ config })
-    const chain = new MockChain({ config })
+    const blockchain = new Blockchain({
+      validateBlocks: false,
+      validateConsensus: false,
+    })
+    const chain = new MockChain({ config, blockchain })
     const serviceConfig = new Config({ loglevel, servers: [server as any], lightserv: true })
     const service = new FullEthereumService({
       config: serviceConfig,

packages/client/test/integration/peerpool.spec.ts

@@ -4,6 +4,7 @@ import { EthProtocol } from '../../lib/net/protocol'
 import { PeerPool } from '../../lib/net/peerpool'
 import MockServer from './mocks/mockserver'
 import MockChain from './mocks/mockchain'
+import Blockchain from '@ethereumjs/blockchain'
 
 tape('[Integration:PeerPool]', async (t) => {
   async function setup(protocols: EthProtocol[] = []): Promise<[MockServer, PeerPool]> {
@@ -57,7 +58,11 @@ tape('[Integration:PeerPool]', async (t) => {
 
   t.test('should handle peer messages', async (t) => {
     const config = new Config({ transports: [], loglevel: 'error' })
-    const chain = new MockChain({ config })
+    const blockchain = new Blockchain({
+      validateBlocks: false,
+      validateConsensus: false,
+    })
+    const chain = new MockChain({ config, blockchain })
     await chain.open()
     const protocols = [
       new EthProtocol({

packages/client/test/integration/util.ts

@@ -2,6 +2,7 @@ import { Config } from '../../lib/config'
 import { FullEthereumService, LightEthereumService } from '../../lib/service'
 import MockServer from './mocks/mockserver'
 import MockChain from './mocks/mockchain'
+import Blockchain from '@ethereumjs/blockchain'
 
 interface SetupOptions {
   location?: string
@@ -20,7 +21,11 @@ export async function setup(
   const config = new Config({ loglevel, syncmode, lightserv })
 
   const server = new MockServer({ config, location })
-  const chain = new MockChain({ config, height })
+  const blockchain = new Blockchain({
+    validateBlocks: false,
+    validateConsensus: false,
+  })
+  const chain = new MockChain({ config, blockchain, height })
 
   const servers = [server] as any
   const serviceConfig = new Config({ loglevel, syncmode, servers, lightserv, minPeers: 1 })

packages/devp2p/src/rlpx/peer.ts

@@ -198,11 +198,14 @@ export class Peer extends EventEmitter {
     if (this._closed) return false
     const msg = Buffer.concat([rlp.encode(code), data])
     const header = this._eciesSession.createHeader(msg.length)
-    if (!header) return
+    if (!header || this._socket.destroyed) return
     this._socket.write(header)
 
     const body = this._eciesSession.createBody(msg)
-    if (!body) return
+    // this._socket.destroyed added here and above to safeguard against
+    // occasional "Cannot call write after a stream was destroyed" errors.
+    // Eventually this can be caught earlier down the line.
+    if (!body || this._socket.destroyed) return
     this._socket.write(body)
     return true
   }