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

CHANGELOG.md

@@ -82,6 +82,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.

config/server_app_options.go

@@ -2,6 +2,7 @@ package config
 
 import (
 	"math"
+	"os"
 	"path/filepath"
 
 	"github.com/holiman/uint256"
@@ -140,6 +141,14 @@ func GetLegacyPoolConfig(appOpts servertypes.AppOptions, logger log.Logger) *leg
 		legacyConfig.Lifetime = lifetime
 	}
 
+	// Set journal path under .evmd data dir and ensure dir exists
+	homeDir := cast.ToString(appOpts.Get(flags.FlagHome))
+	if homeDir != "" {
+		journalDir := filepath.Join(homeDir, "data", "txpool")
+		_ = os.MkdirAll(journalDir, 0o755)
+		legacyConfig.Journal = filepath.Join(journalDir, legacyConfig.Journal)
+	}
+
 	return &legacyConfig
 }
 

mempool/mempool.go

@@ -5,6 +5,7 @@ import (
 	"errors"
 	"fmt"
 	"sync"
+	"time"
 
 	ethtypes "github.com/ethereum/go-ethereum/core/types"
 	"github.com/holiman/uint256"
@@ -14,6 +15,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 +47,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
@@ -130,6 +133,7 @@ func NewExperimentalEVMMempool(
 		// from queued into pending, noting their readiness to be executed.
 		legacyPool.BroadcastTxFn = func(txs []*ethtypes.Transaction) error {
 			logger.Debug("broadcasting EVM transactions", "tx_count", len(txs))
+			fmt.Println(clientCtx)
 			return broadcastEVMTransactions(clientCtx, txConfig, txs)
 		}
 	}
@@ -146,6 +150,21 @@ 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
+		}
+		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 +199,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 +458,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/track_local_txs_test.go

@@ -0,0 +1,4 @@
+// Intentionally left empty; comprehensive TxTracker behavior is tested in
+// package locals where internals are accessible. This file reserved for future
+// integration wiring tests if needed.
+package mempool

mempool/txpool/locals/errors.go

@@ -0,0 +1,30 @@
+package locals
+
+import (
+	"errors"
+
+	"github.com/cosmos/evm/mempool/txpool"
+	"github.com/cosmos/evm/mempool/txpool/legacypool"
+)
+
+// 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.
+func IsTemporaryReject(err error) bool {
+	switch {
+	case errors.Is(err, legacypool.ErrOutOfOrderTxFromDelegated):
+		return true
+	case errors.Is(err, txpool.ErrInflightTxLimitReached):
+		return true
+	case errors.Is(err, legacypool.ErrAuthorityReserved):
+		return true
+	case errors.Is(err, txpool.ErrUnderpriced):
+		return true
+	case errors.Is(err, legacypool.ErrTxPoolOverflow):
+		return true
+	case errors.Is(err, legacypool.ErrFutureReplacePending):
+		return true
+	default:
+		return false
+	}
+}

mempool/txpool/locals/errors_test.go

@@ -0,0 +1,48 @@
+package locals
+
+import (
+	"errors"
+	"testing"
+
+	"github.com/cosmos/evm/mempool/txpool"
+	"github.com/cosmos/evm/mempool/txpool/legacypool"
+)
+
+func TestIsTemporaryReject_PositiveCases(t *testing.T) {
+	cases := []struct {
+		name string
+		err  error
+	}{
+		{name: "delegated out-of-order nonce", err: legacypool.ErrOutOfOrderTxFromDelegated},
+		{name: "inflight tx limit reached", err: txpool.ErrInflightTxLimitReached},
+		{name: "authority reserved", err: legacypool.ErrAuthorityReserved},
+		{name: "underpriced", err: txpool.ErrUnderpriced},
+		{name: "txpool overflow", err: legacypool.ErrTxPoolOverflow},
+		{name: "future replace pending", err: legacypool.ErrFutureReplacePending},
+	}
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			if !IsTemporaryReject(tc.err) {
+				t.Fatalf("expected temporary reject error to be detected, got false: %v", tc.err)
+			}
+		})
+	}
+}
+
+func TestIsTemporaryReject_NegativeCases(t *testing.T) {
+	cases := []struct {
+		name string
+		err  error
+	}{
+		{name: "nil", err: nil},
+		{name: "unrelated", err: errors.New("some unrelated error")},
+		{name: "substring lookalike", err: errors.New("under price threshold")},
+	}
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			if IsTemporaryReject(tc.err) {
+				t.Fatalf("did not expect temporary reject error for: %v", tc.err)
+			}
+		})
+	}
+}