feat(mempool): add timeout for reorg

mempool/iterator.go

@@ -35,17 +35,24 @@ type EVMMempoolIterator struct {
 
 	/** Chain Params **/
 	bondDenom string
-	chainID   *big.Int
 
 	/** Blockchain Access **/
 	blockchain *Blockchain
+	chainID    *big.Int
 }
 
 // NewEVMMempoolIterator creates a new unified iterator over EVM and Cosmos transactions.
 // It combines iterators from both transaction pools and selects transactions based on fee priority.
 // Returns nil if both iterators are empty or nil. The bondDenom parameter specifies the native
 // token denomination for fee comparisons, and chainId is used for EVM transaction conversion.
-func NewEVMMempoolIterator(evmIterator *miner.TransactionsByPriceAndNonce, cosmosIterator mempool.Iterator, logger log.Logger, txConfig client.TxConfig, bondDenom string, chainID *big.Int, blockchain *Blockchain) mempool.Iterator {
+func NewEVMMempoolIterator(
+	evmIterator *miner.TransactionsByPriceAndNonce,
+	cosmosIterator mempool.Iterator,
+	logger log.Logger,
+	txConfig client.TxConfig,
+	bondDenom string,
+	blockchain *Blockchain,
+) mempool.Iterator {
 	// Check if we have any transactions at all
 	hasEVM := evmIterator != nil && !evmIterator.Empty()
 	hasCosmos := cosmosIterator != nil && cosmosIterator.Tx() != nil
@@ -64,8 +71,8 @@ func NewEVMMempoolIterator(evmIterator *miner.TransactionsByPriceAndNonce, cosmo
 		logger:         logger,
 		txConfig:       txConfig,
 		bondDenom:      bondDenom,
-		chainID:        chainID,
 		blockchain:     blockchain,
+		chainID:        blockchain.Config().ChainID,
 	}
 }
 

mempool/mempool.go

@@ -4,7 +4,9 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"math/big"
 	"sync"
+	"time"
 
 	"github.com/ethereum/go-ethereum/common"
 	ethtypes "github.com/ethereum/go-ethereum/core/types"
@@ -57,6 +59,7 @@ type (
 		blockchain    *Blockchain
 		blockGasLimit uint64 // Block gas limit from consensus parameters
 		minTip        *uint256.Int
+		chainID       *big.Int
 
 		/** Verification **/
 		anteHandler sdk.AnteHandler
@@ -192,6 +195,7 @@ func NewExperimentalEVMMempool(
 		minTip:             config.MinTip,
 		anteHandler:        config.AnteHandler,
 		operateExclusively: config.OperateExclusively,
+		chainID:            blockchain.Config().ChainID,
 		reapList:           NewReapList(txEncoder),
 	}
 
@@ -386,21 +390,55 @@ func (m *ExperimentalEVMMempool) ReapNewValidTxs(maxBytes uint64, maxGas uint64)
 
 // Select returns a unified iterator over both EVM and Cosmos transactions.
 // The iterator prioritizes transactions based on their fees and manages proper
-// sequencing. The i parameter contains transaction hashes to exclude from selection.
-func (m *ExperimentalEVMMempool) Select(goCtx context.Context, i [][]byte) sdkmempool.Iterator {
+// sequencing. The txs parameter contains transactions from cometbft consensus.
+func (m *ExperimentalEVMMempool) Select(ctx context.Context, txs [][]byte) sdkmempool.Iterator {
 	m.mtx.Lock()
 	defer m.mtx.Unlock()
-	ctx := sdk.UnwrapSDKContext(goCtx)
 
-	// Wait for the legacypool to Reset at >= blockHeight (this may have
-	// already happened), to ensure all txs in pending pool are valid.
-	m.legacyTxPool.WaitForReorgHeight(ctx, ctx.BlockHeight())
+	return m.selectAfterReorg(ctx, txs)
+}
 
-	evmIterator, cosmosIterator := m.getIterators(goCtx, i)
+// SelectBy iterates through transactions until the provided filter function returns false.
+// It uses the same unified iterator as Select but allows early termination based on
+// custom criteria defined by the filter function.
+func (m *ExperimentalEVMMempool) SelectBy(ctx context.Context, txs [][]byte, filter func(sdk.Tx) bool) {
+	m.mtx.Lock()
+	defer m.mtx.Unlock()
 
-	combinedIterator := NewEVMMempoolIterator(evmIterator, cosmosIterator, m.logger, m.txConfig, m.vmKeeper.GetEvmCoinInfo(ctx).Denom, m.blockchain.Config().ChainID, m.blockchain)
+	iterator := m.selectAfterReorg(ctx, txs)
 
-	return combinedIterator
+	for iterator != nil && filter(iterator.Tx()) {
+		iterator = iterator.Next()
+	}
+}
+
+// selectAfterReorg selects transactions from the EVM and Cosmos mempools after a (partial) reorg has completed.
+// Note that actually txs arg isn't used neither in the evm nor cosmos mempools :harold:
+func (m *ExperimentalEVMMempool) selectAfterReorg(ctx context.Context, txs [][]byte) sdkmempool.Iterator {
+	// max time for waiting for a reorg to complete.
+	// if the reorg doesn't complete in this time, we grab only subset of txs.
+	// this ensures consensus liveness and prevents the mempool from being a bottleneck
+	// during txs congestions and slow re-checks.
+	const maxReorgWait = 300 * time.Millisecond
+
+	var (
+		sdkCtx      = sdk.UnwrapSDKContext(ctx)
+		denom       = m.vmKeeper.GetEvmCoinInfo(sdkCtx).Denom
+		blockHeight = sdkCtx.BlockHeight()
+	)
+
+	m.legacyTxPool.WaitForReorgHeight(ctx, blockHeight, maxReorgWait)
+
+	evmIterator, cosmosIterator := m.getIterators(ctx, txs)
+
+	return NewEVMMempoolIterator(
+		evmIterator,
+		cosmosIterator,
+		m.logger,
+		m.txConfig,
+		denom,
+		m.blockchain,
+	)
 }
 
 // CountTx returns the total number of transactions in both EVM and Cosmos pools.
@@ -501,23 +539,6 @@ func (m *ExperimentalEVMMempool) shouldRemoveFromEVMPool(hash common.Hash, reaso
 	return true
 }
 
-// SelectBy iterates through transactions until the provided filter function returns false.
-// It uses the same unified iterator as Select but allows early termination based on
-// custom criteria defined by the filter function.
-func (m *ExperimentalEVMMempool) SelectBy(goCtx context.Context, i [][]byte, f func(sdk.Tx) bool) {
-	m.mtx.Lock()
-	defer m.mtx.Unlock()
-	ctx := sdk.UnwrapSDKContext(goCtx)
-
-	evmIterator, cosmosIterator := m.getIterators(goCtx, i)
-
-	combinedIterator := NewEVMMempoolIterator(evmIterator, cosmosIterator, m.logger, m.txConfig, m.vmKeeper.GetEvmCoinInfo(ctx).Denom, m.blockchain.Config().ChainID, m.blockchain)
-
-	for combinedIterator != nil && f(combinedIterator.Tx()) {
-		combinedIterator = combinedIterator.Next()
-	}
-}
-
 // SetEventBus sets CometBFT event bus to listen for new block header event.
 func (m *ExperimentalEVMMempool) SetEventBus(eventBus *cmttypes.EventBus) {
 	if m.HasEventBus() {
@@ -576,9 +597,10 @@ func (m *ExperimentalEVMMempool) getEVMMessage(tx sdk.Tx) (*evmtypes.MsgEthereum
 // getIterators prepares iterators over pending EVM and Cosmos transactions.
 // It configures EVM transactions with proper base fee filtering and priority ordering,
 // while setting up the Cosmos iterator with the provided exclusion list.
-func (m *ExperimentalEVMMempool) getIterators(goCtx context.Context, i [][]byte) (*miner.TransactionsByPriceAndNonce, sdkmempool.Iterator) {
+func (m *ExperimentalEVMMempool) getIterators(goCtx context.Context, txs [][]byte) (*miner.TransactionsByPriceAndNonce, sdkmempool.Iterator) {
 	ctx := sdk.UnwrapSDKContext(goCtx)
 	baseFee := m.vmKeeper.GetBaseFee(ctx)
+
 	var baseFeeUint *uint256.Int
 	if baseFee != nil {
 		baseFeeUint = uint256.MustFromBig(baseFee)
@@ -593,10 +615,11 @@ func (m *ExperimentalEVMMempool) getIterators(goCtx context.Context, i [][]byte)
 		OnlyPlainTxs: true,
 		OnlyBlobTxs:  false,
 	}
+
 	evmPendingTxes := m.txPool.Pending(pendingFilter)
-	orderedEVMPendingTxes := miner.NewTransactionsByPriceAndNonce(nil, evmPendingTxes, baseFee)
+	evmIterator := miner.NewTransactionsByPriceAndNonce(nil, evmPendingTxes, baseFee)
 
-	cosmosPendingTxes := m.cosmosPool.Select(ctx, i)
+	cosmosIterator := m.cosmosPool.Select(ctx, txs)
 
-	return orderedEVMPendingTxes, cosmosPendingTxes
+	return evmIterator, cosmosIterator
 }