Refactor TxPool: leaner and simpler (#2973)

* Refactor TxPool: leaner and simpler
* Rewrite test_txpool

Reduce number of tables used, from 5 to 2. Reduce number of files.
If need to modify the price rule or other filters, now is far more easier because only one table to work with(sender/nonce).
And the other table is just a map from txHash to TxItemRef.

Removing transactions from txPool either because of producing new block or syncing became much easier.
Removing expired transactions also simple.
Explicit Tx Pending, Staged, or Packed status is removed. The status of the transactions can be inferred implicitly.
Developer new to TxPool can easily follow the logic.

But the most important is we can revive the test_txpool without dirty trick and remove usage of getCanonicalHead furthermore to prepare for better integration with ForkedChain.

hive_integration/nodocker/engine/engine_env.nim

@@ -92,10 +92,6 @@ proc newEngineEnv*(conf: var NimbusConf, chainFile: string, enableAuth: bool): E
     chain,
     txPool)
 
-  # txPool must be informed of active head
-  # so it can know the latest account state
-  doAssert txPool.smartHead(chain.latestHeader)
-
   var key: JwtSharedKey
   key.fromHex(jwtSecret).isOkOr:
     echo "JWT SECRET ERROR: ", error
@@ -178,14 +174,12 @@ proc peer*(env: EngineEnv): Peer =
 proc getTxsInPool*(env: EngineEnv, txHashes: openArray[common.Hash32]): seq[Transaction] =
   result = newSeqOfCap[Transaction](txHashes.len)
   for txHash in txHashes:
-    let res = env.txPool.getItem(txHash)
-    if res.isErr: continue
-    let item = res.get
-    if item.reject == txInfoOk:
-      result.add item.tx
+    let item = env.txPool.getItem(txHash).valueOr:
+      continue
+    result.add item.tx
 
 proc numTxsInPool*(env: EngineEnv): int =
-  env.txPool.numTxs
+  env.txPool.len
 
 func version*(env: EngineEnv, time: EthTime): Version =
   if env.com.isPragueOrLater(time):

hive_integration/nodocker/engine/tx_sender.nim

@@ -69,6 +69,7 @@ type
     gasPriceOrGasFeeCap*: Opt[GasInt]
     gasTipCap*          : Opt[GasInt]
     gas*                : Opt[GasInt]
+    blobGas*            : Opt[UInt256]
     to*                 : Opt[common.Address]
     value*              : Opt[UInt256]
     data*               : Opt[seq[byte]]
@@ -95,8 +96,8 @@ proc createAccount(idx: int): TestAccount =
     quit(QuitFailure)
   result.address = toAddress(result.key)
 
-proc createAccounts(sender: TxSender) =
-  for i in 0..<TestAccountCount:
+proc createAccounts(sender: TxSender, numAccounts: int) =
+  for i in 0..<numAccounts:
     sender.accounts.add createAccount(i.int)
 
 proc getNextAccount*(sender: TxSender): TestAccount =
@@ -113,14 +114,15 @@ proc getLastNonce(sender: TxSender, address: Address): uint64 =
   sender.nonceMap[address] - 1
 
 proc fillBalance(sender: TxSender, params: NetworkParams) =
+  const balance = UInt256.fromHex("0x123450000000000000000")
   for x in sender.accounts:
     params.genesis.alloc[x.address] = GenesisAccount(
-      balance: UInt256.fromHex("0x123450000000000000000"),
+      balance: balance,
     )
 
-proc new*(_: type TxSender, params: NetworkParams): TxSender =
+proc new*(_: type TxSender, params: NetworkParams, numAccounts = TestAccountCount): TxSender =
   result = TxSender(chainId: params.config.chainId)
-  result.createAccounts()
+  result.createAccounts(numAccounts)
   result.fillBalance(params)
 
 proc getTxType(tc: BaseTx, nonce: uint64): TxType =
@@ -443,6 +445,10 @@ proc customizeTransaction*(sender: TxSender,
       var address: Address
       modTx.to = Opt.some(address)
 
+  if custTx.blobGas.isSome:
+    doAssert(baseTx.txType == TxEip4844)
+    modTx.maxFeePerBlobGas = custTx.blobGas.get
+
   if custTx.signature.isSome:
     let signature = custTx.signature.get
     modTx.V = signature.V

hive_integration/nodocker/graphql/graphql_sim.nim

@@ -85,18 +85,13 @@ proc main() =
     )
     chain   = ForkedChainRef.init(com)
     txPool  = TxPoolRef.new(chain)
-    
+
   discard importRlpBlock(blocksFile, com)
   let ctx = setupGraphqlContext(com, ethNode, txPool)
 
   var stat: SimStat
   let start = getTime()
 
-  # txPool must be informed of active head
-  # so it can know the latest account state
-  # e.g. "sendRawTransaction Nonce too low" case
-  doAssert txPool.smartHead(chain.latestHeader)
-
   for fileName in walkDirRec(
                  caseFolder, yieldFilter = {pcFile,pcLinkToFile}):
     if not fileName.endsWith(".json"):

hive_integration/nodocker/rpc/test_env.nim

@@ -90,10 +90,6 @@ proc setupEnv*(taskPool: Taskpool): TestEnv =
   let chain = ForkedChainRef.init(com)
   let txPool = TxPoolRef.new(chain)
 
-  # txPool must be informed of active head
-  # so it can know the latest account state
-  doAssert txPool.smartHead(chain.latestHeader)
-
   let rpcServer = setupRpcServer(ethCtx, com, ethNode, txPool, conf, chain)
   let rpcClient = newRpcHttpClient()
   waitFor rpcClient.connect("127.0.0.1", Port(8545), false)

nimbus/beacon/api_handler/api_forkchoice.nim

@@ -10,12 +10,13 @@
 import
   std/[typetraits],
   results,
-  ../beacon_engine,
   eth/common/[headers, hashes, times],
   web3/execution_types,
-  ./api_utils,
   chronicles,
-  ../web3_eth_conv
+  ../../core/tx_pool,
+  ../beacon_engine,
+  ../web3_eth_conv,
+  ./api_utils
 
 {.push gcsafe, raises:[CatchableError].}
 
@@ -200,6 +201,7 @@ proc forkchoiceUpdated*(ben: BeaconEngineRef,
       gasUsed = bundle.payload.gasUsed,
       blobGasUsed = bundle.payload.blobGasUsed.get(Quantity(0)),
       id = id.toHex,
+      txPoolLen = ben.txPool.len,
       attrs = attrs
 
     return validFCU(Opt.some(id), blockHash)

nimbus/beacon/api_handler/api_newpayload.nim

@@ -9,13 +9,14 @@
 
 import
   results,
-  ../web3_eth_conv,
+  chronicles,
   eth/common/hashes,
-  ../beacon_engine,
   web3/[execution_types, primitives],
+  ../../core/tx_pool,
+  ../web3_eth_conv,
+  ../beacon_engine,
   ../payload_conv,
-  ./api_utils,
-  chronicles
+  ./api_utils
 
 {.push gcsafe, raises:[CatchableError].}
 
@@ -228,6 +229,8 @@ proc newPayload*(ben: BeaconEngineRef,
     let blockHash = latestValidHash(db, parent, ttd)
     return invalidStatus(blockHash, vres.error())
 
+  ben.txPool.removeNewBlockTxs(blk, Opt.some(blockHash))
+
   info "New payload received and validated",
     number = header.number,
     hash = blockHash.short,

nimbus/beacon/beacon_engine.nim

@@ -126,6 +126,9 @@ func com*(ben: BeaconEngineRef): CommonRef =
 func chain*(ben: BeaconEngineRef): ForkedChainRef =
   ben.txPool.chain
 
+func txPool*(ben: BeaconEngineRef): TxPoolRef =
+  ben.txPool
+
 func get*(ben: BeaconEngineRef, hash: Hash32,
           header: var Header): bool =
   ben.queue.get(hash, header)
@@ -155,10 +158,6 @@ proc generateExecutionBundle*(ben: BeaconEngineRef,
 
     pos.setWithdrawals(attrs)
 
-    if headBlock.blockHash != xp.head.blockHash:
-       # reorg
-       discard xp.smartHead(headBlock)
-
     if pos.timestamp <= headBlock.timestamp:
       return err "timestamp must be strictly later than parent"
 

nimbus/core/chain/forked_chain.nim

@@ -799,3 +799,18 @@ proc isCanonicalAncestor*(c: ForkedChainRef,
   let canonHash = c.db.getBlockHash(blockNumber).valueOr:
     return false
   canonHash == blockHash
+
+iterator txHashInRange*(c: ForkedChainRef, fromHash: Hash32, toHash: Hash32): Hash32 =
+  ## exclude base from iteration, new block produced by txpool
+  ## should not reach base
+  var prevHash = fromHash
+  while prevHash != c.baseHash:
+    c.blocks.withValue(prevHash, item) do:
+      if toHash == prevHash:
+        break
+      for tx in item.blk.transactions:
+        let txHash = rlpHash(tx)
+        yield txHash
+      prevHash = item.blk.header.parentHash
+    do:
+      break