client/engine: speed and resiliency improvements (#1827)

* CLConnectionManager: start on first updateStatus in case PreMerge wasn't reached before merge
* chain: safer closing to not cause db corruption during engine requests
* parse: some improvements and reorganization
* vm/client/engine: add ability to add blocks to blockchain without setting the head. this allows us to do runBlock work in newPayload once, and not again in fcU
* CLConnectionManager: improve logging output
* use ellipsis character
* add space to align with fcu log msg
* add last payload txs count
* add baseFee to last payload log
* fix type doc ref
* add gasUsed to last payload log
* rename PreMerge to MergeForkBlock for increased clarity
use hardfork enums in vm supportedHardforks
* format numbers to locale string, use compact num for gasUsed
update from 5 to 4 chars on each side of hash
* fix tests, use `pass` instead of `ok`
* various fixes, remove toLocaleString, improve short func
* add plural to timeDiffStr if not 1
* Add coverage on `formatNonce`
* Add CLConnectionManager test skeleton and fix `running` method
* Add forkchoice/newpayload tests
* Remove listeners on test
* Tighten up tests to check for specific block
* More connection manager constructor tests
* nits
* remove leading zeros on cumulative gas in eth_getTransactionsReceipt (thanks @cbrzn)
Co-authored-by: acolytec3 <17355484+acolytec3@users.noreply.github.com>

Author: ryanio

packages/client/lib/blockchain/chain.ts

@@ -178,10 +178,7 @@ export class Chain {
    * @returns false if chain is already open, otherwise void
    */
   async open(): Promise<boolean | void> {
-    if (this.opened) {
-      return false
-    }
-
+    if (this.opened) return false
     await this.blockchain.db.open()
     await this.blockchain.initPromise
     this.opened = true
@@ -206,9 +203,7 @@ export class Chain {
    * @returns false if chain is closed, otherwise void
    */
   async close(): Promise<boolean | void> {
-    if (!this.opened) {
-      return false
-    }
+    if (!this.opened) return false
     this.reset()
     await this.blockchain.db.close()
     this.opened = false
@@ -220,9 +215,7 @@ export class Chain {
    * @returns false if chain is closed, otherwise void
    */
   async update(emit = true): Promise<boolean | void> {
-    if (!this.opened) {
-      return false
-    }
+    if (!this.opened) return false
 
     const headers: ChainHeaders = {
       latest: null,
@@ -263,7 +256,7 @@ export class Chain {
    * @returns an array of the blocks
    */
   async getBlocks(block: Buffer | BN, max = 1, skip = 0, reverse = false): Promise<Block[]> {
-    await this.open()
+    if (!this.opened) throw new Error('Chain closed')
     return this.blockchain.getBlocks(block, max, skip, reverse)
   }
 
@@ -273,7 +266,7 @@ export class Chain {
    * @throws if block is not found
    */
   async getBlock(block: Buffer | BN): Promise<Block> {
-    await this.open()
+    if (!this.opened) throw new Error('Chain closed')
     return this.blockchain.getBlock(block)
   }
 
@@ -284,10 +277,9 @@ export class Chain {
    * @returns number of blocks added
    */
   async putBlocks(blocks: Block[], fromEngine = false): Promise<number> {
-    if (blocks.length === 0) {
-      return 0
-    }
-    await this.open()
+    if (!this.opened) throw new Error('Chain closed')
+    if (blocks.length === 0) return 0
+
     let numAdded = 0
     for (const [i, b] of blocks.entries()) {
       if (!fromEngine && this.config.chainCommon.gteHardfork(Hardfork.Merge)) {
@@ -334,10 +326,9 @@ export class Chain {
    * @returns number of headers added
    */
   async putHeaders(headers: BlockHeader[], mergeIncludes = false): Promise<number> {
-    if (headers.length === 0) {
-      return 0
-    }
-    await this.open()
+    if (!this.opened) throw new Error('Chain closed')
+    if (headers.length === 0) return 0
+
     let numAdded = 0
     for (const [i, h] of headers.entries()) {
       if (!mergeIncludes && this.config.chainCommon.gteHardfork(Hardfork.Merge)) {
@@ -363,15 +354,15 @@ export class Chain {
    * Gets the latest header in the canonical chain
    */
   async getLatestHeader(): Promise<BlockHeader> {
-    await this.open()
+    if (!this.opened) throw new Error('Chain closed')
     return this.blockchain.getLatestHeader()
   }
 
   /**
    * Gets the latest block in the canonical chain
    */
   async getLatestBlock(): Promise<Block> {
-    await this.open()
+    if (!this.opened) throw new Error('Chain closed')
     return this.blockchain.getLatestBlock()
   }
 
@@ -382,7 +373,7 @@ export class Chain {
    * @returns the td
    */
   async getTd(hash: Buffer, num: BN): Promise<BN> {
-    await this.open()
+    if (!this.opened) throw new Error('Chain closed')
     return this.blockchain.getTotalDifficulty(hash, num)
   }
 }

packages/client/lib/execution/vmexecution.ts

@@ -1,3 +1,9 @@
+import {
+  DBSetTD,
+  DBSaveLookups,
+  DBSetBlockOrHeader,
+  DBSetHashToNumber,
+} from '@ethereumjs/blockchain/dist/db/helpers'
 import { ConsensusType, Hardfork } from '@ethereumjs/common'
 import VM from '@ethereumjs/vm'
 import { DefaultStateManager } from '@ethereumjs/vm/dist/state'
@@ -6,16 +12,20 @@ import { short } from '../util'
 import { debugCodeReplayBlock } from '../util/debug'
 import { Event } from '../types'
 import { Execution, ExecutionOptions } from './execution'
-import type { Block } from '@ethereumjs/block'
 import { ReceiptsManager } from './receipt'
+import type { Block } from '@ethereumjs/block'
+import type { RunBlockOpts } from '@ethereumjs/vm/dist/runBlock'
+import type { TxReceipt } from '@ethereumjs/vm/dist/types'
 
 export class VMExecution extends Execution {
   public vm: VM
   public hardfork: string = ''
 
   public receiptsManager?: ReceiptsManager
+  private pendingReceipts?: Map<string, TxReceipt[]>
   private vmPromise?: Promise<number | undefined>
 
+  /** Number of maximum blocks to run per iteration of {@link VMExecution.run} */
   private NUM_BLOCKS_PER_ITERATION = 50
 
   /**
@@ -48,6 +58,7 @@ export class VMExecution extends Execution {
         config: this.config,
         metaDB: this.metaDB,
       })
+      this.pendingReceipts = new Map()
     }
   }
 
@@ -66,6 +77,51 @@ export class VMExecution extends Execution {
     }
   }
 
+  /**
+   * Executes the block, runs the necessary verification on it,
+   * and persists the block and the associate state into the database.
+   * The key difference is it won't do the canonical chain updating.
+   * It relies on the additional {@link VMExecution.setHead} call to finalize
+   * the entire procedure.
+   * @param receipts If we built this block, pass the receipts to not need to run the block again
+   */
+  async runWithoutSetHead(opts: RunBlockOpts, receipts?: TxReceipt[]): Promise<void> {
+    const { block } = opts
+    if (receipts === undefined) {
+      const result = await this.vm.runBlock(opts)
+      receipts = result.receipts
+    }
+    if (receipts) {
+      // Save receipts
+      this.pendingReceipts?.set(block.hash().toString('hex'), receipts)
+    }
+    // Bypass updating head by using blockchain db directly
+    const [hash, num] = [block.hash(), block.header.number]
+    const td = (await this.chain.getTd(block.header.parentHash, block.header.number.subn(1))).add(
+      block.header.difficulty
+    )
+    await this.chain.blockchain.dbManager.batch([
+      DBSetTD(td, num, hash),
+      ...DBSetBlockOrHeader(block),
+      DBSetHashToNumber(hash, num),
+      ...DBSaveLookups(hash, num),
+    ])
+  }
+
+  /**
+   * Sets the chain to set the specified new head block.
+   * Should only be used after {@link VMExecution.runWithoutSetHead}
+   */
+  async setHead(block: Block): Promise<void> {
+    await this.chain.putBlocks([block], true)
+    const receipts = this.pendingReceipts?.get(block.hash().toString('hex'))
+    if (receipts) {
+      void this.receiptsManager?.saveReceipts(block, receipts)
+      this.pendingReceipts?.delete(block.hash().toString('hex'))
+    }
+    await this.chain.blockchain.setIteratorHead('vm', block.hash())
+  }
+
   /**
    * Runs the VM execution
    *

packages/client/lib/miner/pendingBlock.ts

@@ -1,5 +1,6 @@
 import { randomBytes } from 'crypto'
 import type VM from '@ethereumjs/vm'
+import type { TxReceipt } from '@ethereumjs/vm/dist/types'
 import type { BlockBuilder } from '@ethereumjs/vm/dist/buildBlock'
 import type { Block, HeaderData } from '@ethereumjs/block'
 import type { TypedTransaction } from '@ethereumjs/tx'
@@ -113,7 +114,7 @@ export class PendingBlock {
   /**
    * Returns the completed block
    */
-  async build(payloadId: Buffer): Promise<Block | void> {
+  async build(payloadId: Buffer): Promise<void | [block: Block, receipts: TxReceipt[]]> {
     const payload = this.pendingPayloads.find((p) => p[0].equals(payloadId))
     if (!payload) {
       return
@@ -165,6 +166,6 @@ export class PendingBlock {
     // Remove from pendingPayloads
     this.pendingPayloads = this.pendingPayloads.filter((p) => !p[0].equals(payloadId))
 
-    return block
+    return [block, builder.transactionReceipts]
   }
 }