feat(mempool): Add TxTracker support for handling temporary rejections

CHANGELOG.md

@@ -103,6 +103,7 @@
 
 ### FEATURES
 
+- [\#646](https://github.com/cosmos/evm/pull/646) Add TxTracker support for tracking priority transactions and handling temporary rejections
 - [\#665](https://github.com/cosmos/evm/pull/665) Add EvmCodec address codec implementation
 - [\#346](https://github.com/cosmos/evm/pull/346) Add eth_createAccessList method and implementation
 - [\#337](https://github.com/cosmos/evm/pull/337) Support state overrides in eth_call.

mempool/mempool.go

@@ -4,7 +4,10 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"os"
+	"path/filepath"
 	"sync"
+	"time"
 
 	ethtypes "github.com/ethereum/go-ethereum/core/types"
 	"github.com/holiman/uint256"
@@ -14,6 +17,7 @@ import (
 	"github.com/cosmos/evm/mempool/miner"
 	"github.com/cosmos/evm/mempool/txpool"
 	"github.com/cosmos/evm/mempool/txpool/legacypool"
+	"github.com/cosmos/evm/mempool/txpool/locals"
 	"github.com/cosmos/evm/rpc/stream"
 	evmtypes "github.com/cosmos/evm/x/vm/types"
 
@@ -45,9 +49,10 @@ type (
 		vmKeeper VMKeeperI
 
 		/** Mempools **/
-		txPool       *txpool.TxPool
-		legacyTxPool *legacypool.LegacyPool
-		cosmosPool   sdkmempool.ExtMempool
+		txPool         *txpool.TxPool
+		legacyTxPool   *legacypool.LegacyPool
+		localTxTracker *locals.TxTracker
+		cosmosPool     sdkmempool.ExtMempool
 
 		/** Utils **/
 		logger        log.Logger
@@ -146,6 +151,27 @@ func NewExperimentalEVMMempool(
 		panic("tx pool should contain only legacypool")
 	}
 
+	var localTxTracker *locals.TxTracker
+
+	if !legacyConfig.NoLocals {
+		rejournal := legacyConfig.Rejournal
+		if rejournal < time.Second {
+			logger.Debug("Sanitizing invalid txpool journal time", "provided", rejournal, "updated", time.Second)
+			rejournal = time.Second
+		}
+		// Ensure journal directory exists before starting the tracker
+		if dir := filepath.Dir(legacyConfig.Journal); dir != "." {
+			if err := os.MkdirAll(dir, 0755); err != nil {
+				logger.Error("failed to create journal directory", "error", err)
+			}
+		}
+		localTxTracker = locals.New(legacyConfig.Journal, rejournal, blockchain.Config(), txPool)
+		err := localTxTracker.Start()
+		if err != nil {
+			return nil
+		}
+	}
+
 	// TODO: move this logic to evmd.createMempoolConfig and set the max tx there
 	// Create Cosmos Mempool from configuration
 	cosmosPoolConfig := config.CosmosPoolConfig
@@ -180,23 +206,33 @@ func NewExperimentalEVMMempool(
 	cosmosPool = sdkmempool.NewPriorityMempool(*cosmosPoolConfig)
 
 	evmMempool := &ExperimentalEVMMempool{
-		vmKeeper:      vmKeeper,
-		txPool:        txPool,
-		legacyTxPool:  txPool.Subpools[0].(*legacypool.LegacyPool),
-		cosmosPool:    cosmosPool,
-		logger:        logger,
-		txConfig:      txConfig,
-		blockchain:    blockchain,
-		blockGasLimit: config.BlockGasLimit,
-		minTip:        config.MinTip,
-		anteHandler:   config.AnteHandler,
+		vmKeeper:       vmKeeper,
+		txPool:         txPool,
+		legacyTxPool:   txPool.Subpools[0].(*legacypool.LegacyPool),
+		localTxTracker: localTxTracker,
+		cosmosPool:     cosmosPool,
+		logger:         logger,
+		txConfig:       txConfig,
+		blockchain:     blockchain,
+		blockGasLimit:  config.BlockGasLimit,
+		minTip:         config.MinTip,
+		anteHandler:    config.AnteHandler,
 	}
 
 	vmKeeper.SetEvmMempool(evmMempool)
 
 	return evmMempool
 }
 
+// TrackLocalTxs tracks transactions as local priority via TxTracker.
+// No-op if local tracking is not initialized.
+func (m *ExperimentalEVMMempool) TrackLocalTxs(txs []*ethtypes.Transaction) {
+	if m == nil || m.localTxTracker == nil || len(txs) == 0 {
+		return
+	}
+	m.localTxTracker.TrackAll(txs)
+}
+
 // GetBlockchain returns the blockchain interface used for chain head event notifications.
 // This is primarily used to notify the mempool when new blocks are finalized.
 func (m *ExperimentalEVMMempool) GetBlockchain() *Blockchain {
@@ -429,6 +465,10 @@ func (m *ExperimentalEVMMempool) Close() error {
 		errs = append(errs, fmt.Errorf("failed to close txpool: %w", err))
 	}
 
+	if err := m.localTxTracker.Stop(); err != nil {
+		errs = append(errs, fmt.Errorf("failed to close localTxTracker: %w", err))
+	}
+
 	return errors.Join(errs...)
 }
 

mempool/txpool/legacypool/legacypool.go

@@ -18,7 +18,6 @@
 package legacypool
 
 import (
-	"errors"
 	"maps"
 	"math/big"
 	"slices"
@@ -31,6 +30,7 @@ import (
 	"github.com/ethereum/go-ethereum/common/prque"
 	"github.com/ethereum/go-ethereum/consensus/misc/eip1559"
 	"github.com/ethereum/go-ethereum/core"
+	"github.com/ethereum/go-ethereum/core/txpool/legacypool"
 	"github.com/ethereum/go-ethereum/core/types"
 	"github.com/ethereum/go-ethereum/core/vm"
 	"github.com/ethereum/go-ethereum/crypto/kzg4844"
@@ -58,25 +58,6 @@ const (
 	txMaxSize = 4 * txSlotSize // 128KB
 )
 
-var (
-	// ErrTxPoolOverflow is returned if the transaction pool is full and can't accept
-	// another remote transaction.
-	ErrTxPoolOverflow = errors.New("txpool is full")
-
-	// ErrOutOfOrderTxFromDelegated is returned when the transaction with gapped
-	// nonce received from the accounts with delegation or pending delegation.
-	ErrOutOfOrderTxFromDelegated = errors.New("gapped-nonce tx from delegated accounts")
-
-	// ErrAuthorityReserved is returned if a transaction has an authorization
-	// signed by an address which already has in-flight transactions known to the
-	// pool.
-	ErrAuthorityReserved = errors.New("authority already reserved")
-
-	// ErrFutureReplacePending is returned if a future transaction replaces a pending
-	// one. Future transactions should only be able to replace other future transactions.
-	ErrFutureReplacePending = errors.New("future transaction tries to replace pending")
-)
-
 var (
 	evictionInterval    = time.Minute     // Time interval to check for evictable transactions
 	statsReportInterval = 8 * time.Second // Time interval to report transaction pool stats
@@ -135,43 +116,49 @@ type BlockChain interface {
 	StateAt(root common.Hash) (vm.StateDB, error)
 }
 
-// Config are the configuration parameters of the transaction pool.
+// Config defines the configuration for the EVM mempool transaction pool.
 type Config struct {
-	Locals    []common.Address // Addresses that should be treated by default as local
-	NoLocals  bool             // Whether local transaction handling should be disabled
-	Journal   string           // Journal of local transactions to survive node restarts
-	Rejournal time.Duration    // Time interval to regenerate the local transaction journal
-
-	PriceLimit uint64 // Minimum gas price to enforce for acceptance into the pool
-	PriceBump  uint64 // Minimum price bump percentage to replace an already existing transaction (nonce)
-
-	AccountSlots uint64 // Number of executable transaction slots guaranteed per account
-	GlobalSlots  uint64 // Maximum number of executable transaction slots for all accounts
-	AccountQueue uint64 // Maximum number of non-executable transaction slots permitted per account
-	GlobalQueue  uint64 // Maximum number of non-executable transaction slots for all accounts
-
-	Lifetime time.Duration // Maximum amount of time non-executable transaction are queued
-}
-
-// DefaultConfig contains the default configurations for the transaction pool.
+	// PriceLimit is the minimum gas price to enforce for acceptance into the pool
+	PriceLimit uint64 `mapstructure:"price-limit"`
+	// PriceBump is the minimum price bump percentage to replace an already existing transaction (nonce)
+	PriceBump uint64 `mapstructure:"price-bump"`
+	// AccountSlots is the number of executable transaction slots guaranteed per account
+	AccountSlots uint64 `mapstructure:"account-slots"`
+	// GlobalSlots is the maximum number of executable transaction slots for all accounts
+	GlobalSlots uint64 `mapstructure:"global-slots"`
+	// AccountQueue is the maximum number of non-executable transaction slots permitted per account
+	AccountQueue uint64 `mapstructure:"account-queue"`
+	// GlobalQueue is the maximum number of non-executable transaction slots for all accounts
+	GlobalQueue uint64 `mapstructure:"global-queue"`
+	// Lifetime is the maximum amount of time non-executable transaction are queued
+	Lifetime time.Duration `mapstructure:"lifetime"`
+	// Locals is the set of addresses that should be treated by default as local
+	Locals []string `mapstructure:"locals"`
+	// NoLocals disables local transaction handling, exempting local accounts from pricing and acceptance
+	NoLocals bool `mapstructure:"no-locals"`
+	// Journal is the path to the local transaction journal file
+	Journal string `mapstructure:"journal"`
+	// Rejournal is the time interval to regenerate the local transaction journal
+	Rejournal time.Duration `mapstructure:"rejournal"`
+}
+
+// DefaultConfig returns the default mempool configuration
 var DefaultConfig = Config{
-	Journal:   "transactions.rlp",
-	Rejournal: time.Hour,
-
-	PriceLimit: 1,
-	PriceBump:  10,
-
-	AccountSlots: 16,
-	GlobalSlots:  4096 + 1024, // urgent + floating queue capacity with 4:1 ratio
-	AccountQueue: 64,
-	GlobalQueue:  1024,
-
-	Lifetime: 3 * time.Hour,
-}
-
-// sanitize checks the provided user configurations and changes anything that's
-// unreasonable or unworkable.
-func (config *Config) sanitize() Config {
+	PriceLimit:   1,                  // Minimum gas price of 1 wei
+	PriceBump:    10,                 // 10% price bump to replace transaction
+	AccountSlots: 16,                 // 16 executable transaction slots per account
+	GlobalSlots:  5120,               // 4096 + 1024 = 5120 global executable slots
+	AccountQueue: 64,                 // 64 non-executable transaction slots per account
+	GlobalQueue:  1024,               // 1024 global non-executable slots
+	Lifetime:     3 * time.Hour,      // 3 hour lifetime for queued transactions
+	Locals:       []string{},         // No local addresses by default
+	NoLocals:     false,              // Local transaction handling enabled by default
+	Journal:      "transactions.rlp", // Default journal filename
+	Rejournal:    time.Hour,          // Regenerate journal every hour
+}
+
+// Sanitize checks the provided user configurations and changes anything that's unreasonable or unworkable.
+func (config *Config) Sanitize() Config {
 	conf := *config
 	if conf.PriceLimit < 1 {
 		log.Warn("Sanitizing invalid txpool price limit", "provided", conf.PriceLimit, "updated", DefaultConfig.PriceLimit)
@@ -266,7 +253,7 @@ type txpoolResetRequest struct {
 // transactions from the network.
 func New(config Config, chain BlockChain) *LegacyPool {
 	// Sanitize the input to ensure no vulnerable gas prices are set
-	config = (&config).sanitize()
+	config = config.Sanitize()
 
 	// Create the transaction pool with its initial settings
 	pool := &LegacyPool{
@@ -623,7 +610,7 @@ func (pool *LegacyPool) checkDelegationLimit(tx *types.Transaction) error {
 	if pending == nil {
 		// Transaction with gapped nonce is not supported for delegated accounts
 		if pool.pendingNonces.get(from) != tx.Nonce() {
-			return ErrOutOfOrderTxFromDelegated
+			return legacypool.ErrOutOfOrderTxFromDelegated
 		}
 		return nil
 	}
@@ -654,7 +641,7 @@ func (pool *LegacyPool) validateAuth(tx *types.Transaction) error {
 				count += queue.Len()
 			}
 			if count > 1 {
-				return ErrAuthorityReserved
+				return legacypool.ErrAuthorityReserved
 			}
 			// Because there is no exclusive lock held between different subpools
 			// when processing transactions, the SetCode transaction may be accepted
@@ -665,7 +652,7 @@ func (pool *LegacyPool) validateAuth(tx *types.Transaction) error {
 			// that attackers cannot easily stack a SetCode transaction when the sender
 			// is reserved by other pools.
 			if pool.reserver.Has(auth) {
-				return ErrAuthorityReserved
+				return legacypool.ErrAuthorityReserved
 			}
 		}
 	}
@@ -730,7 +717,7 @@ func (pool *LegacyPool) add(tx *types.Transaction) (replaced bool, err error) {
 		// replacements to 25% of the slots
 		if pool.changesSinceReorg > int(pool.config.GlobalSlots/4) {
 			throttleTxMeter.Mark(1)
-			return false, ErrTxPoolOverflow
+			return false, legacypool.ErrTxPoolOverflow
 		}
 
 		// New transaction is better than our worse ones, make room for it.
@@ -741,7 +728,7 @@ func (pool *LegacyPool) add(tx *types.Transaction) (replaced bool, err error) {
 		if !success {
 			log.Trace("Discarding overflown transaction", "hash", hash)
 			overflowedTxMeter.Mark(1)
-			return false, ErrTxPoolOverflow
+			return false, legacypool.ErrTxPoolOverflow
 		}
 
 		// If the new transaction is a future transaction it should never churn pending transactions
@@ -760,7 +747,7 @@ func (pool *LegacyPool) add(tx *types.Transaction) (replaced bool, err error) {
 					pool.priced.Put(dropTx)
 				}
 				log.Trace("Discarding future transaction replacing pending tx", "hash", hash)
-				return false, ErrFutureReplacePending
+				return false, legacypool.ErrFutureReplacePending
 			}
 		}
 

mempool/txpool/legacypool/legacypool_test.go

@@ -33,6 +33,7 @@ import (
 	"github.com/ethereum/go-ethereum/core"
 	"github.com/ethereum/go-ethereum/core/state"
 	"github.com/ethereum/go-ethereum/core/tracing"
+	"github.com/ethereum/go-ethereum/core/txpool/legacypool"
 	"github.com/ethereum/go-ethereum/core/types"
 	"github.com/ethereum/go-ethereum/core/vm"
 	"github.com/ethereum/go-ethereum/crypto"
@@ -1696,8 +1697,8 @@ func TestUnderpricing(t *testing.T) {
 		t.Fatalf("failed to add well priced transaction: %v", err)
 	}
 	// Ensure that replacing a pending transaction with a future transaction fails
-	if err := pool.addRemoteSync(pricedTransaction(5, 100000, big.NewInt(6), keys[1])); !errors.Is(err, ErrFutureReplacePending) {
-		t.Fatalf("adding future replace transaction error mismatch: have %v, want %v", err, ErrFutureReplacePending)
+	if err := pool.addRemoteSync(pricedTransaction(5, 100000, big.NewInt(6), keys[1])); !errors.Is(err, legacypool.ErrFutureReplacePending) {
+		t.Fatalf("adding future replace transaction error mismatch: have %v, want %v", err, legacypool.ErrFutureReplacePending)
 	}
 	pending, queued = pool.Stats()
 	if pending != 4 {
@@ -2297,8 +2298,8 @@ func TestSetCodeTransactions(t *testing.T) {
 				statedb.SetCode(aa, []byte{byte(vm.ADDRESS), byte(vm.PUSH0), byte(vm.SSTORE)})
 
 				// Send gapped transaction, it should be rejected.
-				if err := pool.addRemoteSync(pricedTransaction(2, 100000, big.NewInt(1), keyA)); !errors.Is(err, ErrOutOfOrderTxFromDelegated) {
-					t.Fatalf("%s: error mismatch: want %v, have %v", name, ErrOutOfOrderTxFromDelegated, err)
+				if err := pool.addRemoteSync(pricedTransaction(2, 100000, big.NewInt(1), keyA)); !errors.Is(err, legacypool.ErrOutOfOrderTxFromDelegated) {
+					t.Fatalf("%s: error mismatch: want %v, have %v", name, legacypool.ErrOutOfOrderTxFromDelegated, err)
 				}
 				// Send transactions. First is accepted, second is rejected.
 				if err := pool.addRemoteSync(pricedTransaction(0, 100000, big.NewInt(1), keyA)); err != nil {
@@ -2377,8 +2378,8 @@ func TestSetCodeTransactions(t *testing.T) {
 					t.Fatalf("%s: failed to add with pending delegation: %v", name, err)
 				}
 				// Delegation rejected since two txs are already in-flight.
-				if err := pool.addRemoteSync(setCodeTx(0, keyA, []unsignedAuth{{0, keyB}})); !errors.Is(err, ErrAuthorityReserved) {
-					t.Fatalf("%s: error mismatch: want %v, have %v", name, ErrAuthorityReserved, err)
+				if err := pool.addRemoteSync(setCodeTx(0, keyA, []unsignedAuth{{0, keyB}})); !errors.Is(err, legacypool.ErrAuthorityReserved) {
+					t.Fatalf("%s: error mismatch: want %v, have %v", name, legacypool.ErrAuthorityReserved, err)
 				}
 			},
 		},

mempool/txpool/locals/errors.go

@@ -0,0 +1,37 @@
+package locals
+
+import (
+	"errors"
+	"strings"
+
+	"github.com/cosmos/evm/mempool/txpool"
+	"github.com/ethereum/go-ethereum/core/txpool/legacypool"
+)
+
+var (
+	// ErrNonceGap is returned if the tx nonce is higher than the account nonce.
+	// This is a duplicate of mempool.ErrNonceGap to avoid import cycle.
+	ErrNonceGap = errors.New("tx nonce is higher than account nonce")
+)
+
+// IsTemporaryReject determines whether the given error indicates a temporary reason to reject a
+// transaction from being included in the txpool. The result may change if the txpool's state changes later.
+// We use strings.Contains instead of errors.Is because we are passing in rawLog errors.
+func IsTemporaryReject(err error) bool {
+	if err == nil {
+		return false
+	}
+
+	switch {
+	case strings.Contains(err.Error(), legacypool.ErrOutOfOrderTxFromDelegated.Error()),
+		strings.Contains(err.Error(), txpool.ErrInflightTxLimitReached.Error()),
+		strings.Contains(err.Error(), legacypool.ErrAuthorityReserved.Error()),
+		strings.Contains(err.Error(), txpool.ErrUnderpriced.Error()),
+		strings.Contains(err.Error(), legacypool.ErrTxPoolOverflow.Error()),
+		strings.Contains(err.Error(), legacypool.ErrFutureReplacePending.Error()),
+		strings.Contains(err.Error(), ErrNonceGap.Error()):
+		return true
+	}
+
+	return false
+}