client: move TxPool to FullEthereumService (#1853)

* move txpool from FullSync to FullEthereumService
remove execution from client class
* move execution init to service
* use pool peer count directly now
* nit (dedupe `any` cast)

Author: ryanio

packages/client/bin/cli.ts

@@ -15,6 +15,7 @@ import { Config, DataDirectory } from '../lib/config'
 import { Logger, getLogger } from '../lib/logging'
 import { startRPCServers, helprpc } from './startRpc'
 import type { Chain as IChain, GenesisState } from '@ethereumjs/common/dist/types'
+import type { FullEthereumService } from '../lib/service'
 const level = require('level')
 const yargs = require('yargs/yargs')
 const { hideBin } = require('yargs/helpers')
@@ -309,7 +310,9 @@ async function executeBlocks(client: EthereumClient) {
     )
     process.exit()
   }
-  await client.executeBlocks(first, last, txHashes)
+  const { execution } = client.services.find((s) => s.name === 'eth') as FullEthereumService
+  if (!execution) throw new Error('executeBlocks requires execution')
+  await execution.executeBlocks(first, last, txHashes)
 }
 
 /**

packages/client/lib/client.ts

@@ -1,10 +1,8 @@
-import { bufferToHex } from 'ethereumjs-util'
 import { version as packageVersion } from '../package.json'
 import { MultiaddrLike } from './types'
 import { Config, SyncMode } from './config'
 import { FullEthereumService, LightEthereumService } from './service'
 import { Event } from './types'
-import { VMExecution } from './execution'
 import { Chain } from './blockchain'
 // eslint-disable-next-line implicit-dependencies/no-implicit
 import type { LevelUp } from 'levelup'
@@ -57,7 +55,6 @@ export default class EthereumClient {
   public config: Config
   public chain: Chain
   public services: (FullEthereumService | LightEthereumService)[]
-  public execution: VMExecution | undefined
 
   public opened: boolean
   public started: boolean
@@ -70,20 +67,13 @@ export default class EthereumClient {
     this.chain = new Chain(options)
 
     if (this.config.syncmode === SyncMode.Full) {
-      this.execution = new VMExecution({
-        config: options.config,
-        stateDB: options.stateDB,
-        metaDB: options.metaDB,
-        chain: this.chain,
-      })
       this.services = [
         new FullEthereumService({
           config: this.config,
           chainDB: options.chainDB,
           stateDB: options.stateDB,
           metaDB: options.metaDB,
           chain: this.chain,
-          execution: this.execution,
         }),
       ]
     } else {
@@ -123,7 +113,6 @@ export default class EthereumClient {
       this.config.logger.info(`Synchronized blockchain at height=${height}`)
     })
 
-    await this.execution?.open()
     await Promise.all(this.services.map((s) => s.open()))
 
     this.opened = true
@@ -152,7 +141,6 @@ export default class EthereumClient {
       return false
     }
     this.config.events.emit(Event.CLIENT_SHUTDOWN)
-    await this.execution?.stop()
     await Promise.all(this.services.map((s) => s.stop()))
     await Promise.all(this.config.servers.map((s) => s.stop()))
     this.started = false
@@ -173,62 +161,4 @@ export default class EthereumClient {
   server(name: string) {
     return this.config.servers.find((s) => s.name === name)
   }
-
-  /**
-   * Execute a range of blocks on a copy of the VM
-   * without changing any chain or client state
-   *
-   * Possible input formats:
-   *
-   * - Single block, '5'
-   * - Range of blocks, '5-10'
-   *
-   */
-  async executeBlocks(first: number, last: number, txHashes: string[]) {
-    this.config.logger.info('Preparing for block execution (debug mode, no services started)...')
-    if (!this.execution) throw new Error('executeBlocks requires execution')
-    const vm = this.execution.vm.copy()
-
-    for (let blockNumber = first; blockNumber <= last; blockNumber++) {
-      const block = await vm.blockchain.getBlock(blockNumber)
-      const parentBlock = await vm.blockchain.getBlock(block.header.parentHash)
-
-      // Set the correct state root
-      await vm.stateManager.setStateRoot(parentBlock.header.stateRoot)
-
-      const td = await vm.blockchain.getTotalDifficulty(block.header.parentHash)
-      vm._common.setHardforkByBlockNumber(blockNumber, td)
-
-      if (txHashes.length === 0) {
-        const res = await vm.runBlock({ block })
-        this.config.logger.info(
-          `Executed block num=${blockNumber} hash=0x${block.hash().toString('hex')} txs=${
-            block.transactions.length
-          } gasUsed=${res.gasUsed} `
-        )
-      } else {
-        let count = 0
-        // Special verbose tx execution mode triggered by BLOCK_NUMBER[*]
-        // Useful e.g. to trace slow txs
-        const allTxs = txHashes.length === 1 && txHashes[0] === '*' ? true : false
-        for (const tx of block.transactions) {
-          const txHash = bufferToHex(tx.hash())
-          if (allTxs || txHashes.includes(txHash)) {
-            const res = await vm.runTx({ block, tx })
-            this.config.logger.info(
-              `Executed tx hash=${txHash} gasUsed=${res.gasUsed} from block num=${blockNumber}`
-            )
-            count += 1
-          }
-        }
-        if (count === 0) {
-          if (!allTxs) {
-            this.config.logger.warn(`Block number ${first} contains no txs with provided hashes`)
-          } else {
-            this.config.logger.info(`Block has 0 transactions (no execution)`)
-          }
-        }
-      }
-    }
-  }
 }

packages/client/lib/config.ts

@@ -1,7 +1,7 @@
 import Common, { Hardfork } from '@ethereumjs/common'
 import VM from '@ethereumjs/vm'
 import { genPrivateKey } from '@ethereumjs/devp2p'
-import { Address } from 'ethereumjs-util'
+import { Address, BN } from 'ethereumjs-util'
 import { Multiaddr } from 'multiaddr'
 import { Logger, getLogger } from './logging'
 import { Libp2pServer, RlpxServer } from './net/server'
@@ -253,6 +253,8 @@ export class Config {
 
   public synchronized: boolean
   public lastSyncDate: number
+  /** Best known block height */
+  public syncTargetHeight?: BN
 
   public readonly chainCommon: Common
   public readonly execCommon: Common

packages/client/lib/execution/vmexecution.ts

@@ -6,6 +6,7 @@ import {
 } from '@ethereumjs/blockchain/dist/db/helpers'
 import { ConsensusType, Hardfork } from '@ethereumjs/common'
 import VM from '@ethereumjs/vm'
+import { bufferToHex } from 'ethereumjs-util'
 import { DefaultStateManager } from '@ethereumjs/vm/dist/state'
 import { SecureTrie as Trie } from 'merkle-patricia-tree'
 import { short } from '../util'
@@ -292,4 +293,60 @@ export class VMExecution extends Execution {
     await super.stop()
     return true
   }
+
+  /**
+   * Execute a range of blocks on a copy of the VM
+   * without changing any chain or client state
+   *
+   * Possible input formats:
+   *
+   * - Single block, '5'
+   * - Range of blocks, '5-10'
+   */
+  async executeBlocks(first: number, last: number, txHashes: string[]) {
+    this.config.logger.info('Preparing for block execution (debug mode, no services started)...')
+    const vm = this.vm.copy()
+
+    for (let blockNumber = first; blockNumber <= last; blockNumber++) {
+      const block = await vm.blockchain.getBlock(blockNumber)
+      const parentBlock = await vm.blockchain.getBlock(block.header.parentHash)
+
+      // Set the correct state root
+      await vm.stateManager.setStateRoot(parentBlock.header.stateRoot)
+
+      const td = await vm.blockchain.getTotalDifficulty(block.header.parentHash)
+      vm._common.setHardforkByBlockNumber(blockNumber, td)
+
+      if (txHashes.length === 0) {
+        const res = await vm.runBlock({ block })
+        this.config.logger.info(
+          `Executed block num=${blockNumber} hash=0x${block.hash().toString('hex')} txs=${
+            block.transactions.length
+          } gasUsed=${res.gasUsed} `
+        )
+      } else {
+        let count = 0
+        // Special verbose tx execution mode triggered by BLOCK_NUMBER[*]
+        // Useful e.g. to trace slow txs
+        const allTxs = txHashes.length === 1 && txHashes[0] === '*' ? true : false
+        for (const tx of block.transactions) {
+          const txHash = bufferToHex(tx.hash())
+          if (allTxs || txHashes.includes(txHash)) {
+            const res = await vm.runTx({ block, tx })
+            this.config.logger.info(
+              `Executed tx hash=${txHash} gasUsed=${res.gasUsed} from block num=${blockNumber}`
+            )
+            count += 1
+          }
+        }
+        if (count === 0) {
+          if (!allTxs) {
+            this.config.logger.warn(`Block number ${first} contains no txs with provided hashes`)
+          } else {
+            this.config.logger.info(`Block has 0 transactions (no execution)`)
+          }
+        }
+      }
+    }
+  }
 }

packages/client/lib/miner/miner.ts

@@ -4,7 +4,7 @@ import { BN } from 'ethereumjs-util'
 import { ConsensusType, Hardfork } from '@ethereumjs/common'
 import { Event } from '../types'
 import { Config } from '../config'
-import { FullSynchronizer } from '../sync'
+import { FullEthereumService } from '../service'
 import { VMExecution } from '../execution'
 
 const level = require('level-mem')
@@ -13,9 +13,8 @@ export interface MinerOptions {
   /* Config */
   config: Config
 
-  /* FullSynchronizer */
-  synchronizer: FullSynchronizer
-  execution: VMExecution
+  /* FullEthereumService */
+  service: FullEthereumService
 }
 
 /**
@@ -31,7 +30,7 @@ export class Miner {
   private _nextAssemblyTimeoutId: NodeJS.Timeout | undefined /* global NodeJS */
   private _boundChainUpdatedHandler: (() => void) | undefined
   private config: Config
-  private synchronizer: FullSynchronizer
+  private service: FullEthereumService
   private execution: VMExecution
   private assembling: boolean
   private period: number
@@ -46,8 +45,8 @@ export class Miner {
    */
   constructor(options: MinerOptions) {
     this.config = options.config
-    this.synchronizer = options.synchronizer
-    this.execution = options.execution
+    this.service = options.service
+    this.execution = this.service.execution
     this.running = false
     this.assembling = false
     this.period = (this.config.chainCommon.consensusConfig().period ?? this.DEFAULT_PERIOD) * 1000 // defined in ms for setTimeout use
@@ -61,7 +60,7 @@ export class Miner {
    * Convenience alias to return the latest block in the blockchain
    */
   private latestBlockHeader(): BlockHeader {
-    return (this.synchronizer as any).chain.headers.latest
+    return this.service.chain.headers.latest!
   }
 
   /**
@@ -86,7 +85,7 @@ export class Miner {
       // EIP-225 spec: If the signer is out-of-turn,
       // delay signing by rand(SIGNER_COUNT * 500ms)
       const [signerAddress] = this.config.accounts[0]
-      const { blockchain } = (this.synchronizer as any).chain
+      const { blockchain } = this.service.chain
       const inTurn = await blockchain.cliqueSignerInTurn(signerAddress)
       if (!inTurn) {
         const signerCount = blockchain.cliqueActiveSigners().length
@@ -175,7 +174,7 @@ export class Miner {
     _boundSetInterruptHandler = setInterrupt.bind(this)
     this.config.events.once(Event.CHAIN_UPDATED, _boundSetInterruptHandler)
 
-    const parentBlock = (this.synchronizer as any).chain.blocks.latest
+    const parentBlock = this.service.chain.blocks.latest!
     const number = parentBlock.header.number.addn(1)
     let { gasLimit } = parentBlock.header
 
@@ -186,7 +185,7 @@ export class Miner {
         { number },
         { common: this.config.chainCommon, cliqueSigner }
       )
-      if ((this.synchronizer as any).chain.blockchain.cliqueCheckRecentlySigned(header)) {
+      if ((this.service.chain.blockchain as any).cliqueCheckRecentlySigned(header)) {
         this.config.logger.info(`Miner: We have too recently signed, waiting for next block`)
         this.assembling = false
         return
@@ -258,10 +257,7 @@ export class Miner {
       },
     })
 
-    const txs = await this.synchronizer.txPool.txsByPriceAndNonce(
-      vmCopy.stateManager,
-      baseFeePerGas
-    )
+    const txs = await this.service.txPool.txsByPriceAndNonce(baseFeePerGas)
     this.config.logger.info(
       `Miner: Assembling block from ${txs.length} eligible txs ${
         baseFeePerGas ? `(baseFee: ${baseFeePerGas})` : ''
@@ -304,9 +300,9 @@ export class Miner {
     this.assembling = false
     if (interrupt) return
     // Put block in blockchain
-    await this.synchronizer.handleNewBlock(block)
+    await this.service.synchronizer.handleNewBlock(block)
     // Remove included txs from TxPool
-    this.synchronizer.txPool.removeNewBlockTxs([block])
+    this.service.txPool.removeNewBlockTxs([block])
     this.config.events.removeListener(Event.CHAIN_UPDATED, _boundSetInterruptHandler)
   }
 

packages/client/lib/miner/pendingBlock.ts

@@ -4,7 +4,7 @@ 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'
-import type { TxPool } from '../sync/txpool'
+import type { TxPool } from '../service/txpool'
 import type { Config } from '../config'
 
 interface PendingBlockOpts {
@@ -67,7 +67,7 @@ export class PendingBlock {
     this.pendingPayloads.push([payloadId, builder])
 
     // Add current txs in pool
-    const txs = await this.txPool.txsByPriceAndNonce(vm.stateManager, baseFeePerGas)
+    const txs = await this.txPool.txsByPriceAndNonce(baseFeePerGas)
     this.config.logger.info(
       `Pending: Assembling block from ${txs.length} eligible txs (baseFee: ${baseFeePerGas})`
     )
@@ -123,10 +123,7 @@ export class PendingBlock {
 
     // Add new txs that the pool received
     const txs = (
-      await this.txPool.txsByPriceAndNonce(
-        (builder as any).vm.stateManager,
-        (builder as any).headerData.baseFeePerGas
-      )
+      await this.txPool.txsByPriceAndNonce((builder as any).headerData.baseFeePerGas)
     ).filter(
       (tx) =>
         !(builder as any).transactions.some((t: TypedTransaction) => t.hash().equals(tx.hash()))

packages/client/lib/net/protocol/ethprotocol.ts

@@ -99,7 +99,7 @@ export class EthProtocol extends Protocol {
       decode: ([txs]: [Buffer[]]) => {
         // TODO: add proper Common instance (problem: service not accesible)
         //const common = this.config.chainCommon.copy()
-        //common.setHardforkByBlockNumber(this.service.synchronizer.syncTargetHeight, this.chain.headers.td)
+        //common.setHardforkByBlockNumber(this.config.syncTargetHeight, this.chain.headers.td)
         return txs.map((txData) => TransactionFactory.fromBlockBodyData(txData))
       },
     },
@@ -210,7 +210,7 @@ export class EthProtocol extends Protocol {
         new BN(reqId),
         // TODO: add proper Common instance (problem: service not accesible)
         //const common = this.config.chainCommon.copy()
-        //common.setHardforkByBlockNumber(this.service.synchronizer.syncTargetHeight)
+        //common.setHardforkByBlockNumber(this.config.syncTargetHeight)
         txs.map((txData) => TransactionFactory.fromBlockBodyData(txData)),
       ],
     },