core/txpool: reject stale transaction for local tracking (#31473)

Fixes https://github.com/ethereum/go-ethereum/issues/31451

core/txpool/locals/tx_tracker_test.go

@@ -0,0 +1,179 @@
+// Copyright 2025 The go-ethereum Authors
+// This file is part of the go-ethereum library.
+//
+// The go-ethereum library is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Lesser General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// The go-ethereum library is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU Lesser General Public License for more details.
+//
+// You should have received a copy of the GNU Lesser General Public License
+// along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
+
+package locals
+
+import (
+	"errors"
+	"math/big"
+	"testing"
+	"time"
+
+	"github.com/ethereum/go-ethereum/common"
+	"github.com/ethereum/go-ethereum/consensus/ethash"
+	"github.com/ethereum/go-ethereum/core"
+	"github.com/ethereum/go-ethereum/core/rawdb"
+	"github.com/ethereum/go-ethereum/core/txpool"
+	"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"
+	"github.com/ethereum/go-ethereum/ethdb"
+	"github.com/ethereum/go-ethereum/params"
+)
+
+var (
+	key, _  = crypto.HexToECDSA("b71c71a67e1177ad4e901695e1b4b9ee17ae16c6668d313eac2f96dbcda3f291")
+	address = crypto.PubkeyToAddress(key.PublicKey)
+	funds   = big.NewInt(1000000000000000)
+	gspec   = &core.Genesis{
+		Config: params.TestChainConfig,
+		Alloc: types.GenesisAlloc{
+			address: {Balance: funds},
+		},
+		BaseFee: big.NewInt(params.InitialBaseFee),
+	}
+	signer = types.LatestSigner(gspec.Config)
+)
+
+type testEnv struct {
+	chain   *core.BlockChain
+	pool    *txpool.TxPool
+	tracker *TxTracker
+	genDb   ethdb.Database
+}
+
+func newTestEnv(t *testing.T, n int, gasTip uint64, journal string) *testEnv {
+	genDb, blocks, _ := core.GenerateChainWithGenesis(gspec, ethash.NewFaker(), n, func(i int, gen *core.BlockGen) {
+		tx, err := types.SignTx(types.NewTransaction(gen.TxNonce(address), common.Address{0x00}, big.NewInt(1000), params.TxGas, gen.BaseFee(), nil), signer, key)
+		if err != nil {
+			panic(err)
+		}
+		gen.AddTx(tx)
+	})
+
+	db := rawdb.NewMemoryDatabase()
+	chain, _ := core.NewBlockChain(db, nil, gspec, nil, ethash.NewFaker(), vm.Config{}, nil)
+
+	legacyPool := legacypool.New(legacypool.DefaultConfig, chain)
+	pool, err := txpool.New(gasTip, chain, []txpool.SubPool{legacyPool})
+	if err != nil {
+		t.Fatalf("Failed to create tx pool: %v", err)
+	}
+	if n, err := chain.InsertChain(blocks); err != nil {
+		t.Fatalf("Failed to process block %d: %v", n, err)
+	}
+	if err := pool.Sync(); err != nil {
+		t.Fatalf("Failed to sync the txpool, %v", err)
+	}
+	return &testEnv{
+		chain:   chain,
+		pool:    pool,
+		tracker: New(journal, time.Minute, gspec.Config, pool),
+		genDb:   genDb,
+	}
+}
+
+func (env *testEnv) close() {
+	env.chain.Stop()
+}
+
+func (env *testEnv) setGasTip(gasTip uint64) {
+	env.pool.SetGasTip(new(big.Int).SetUint64(gasTip))
+}
+
+func (env *testEnv) makeTx(nonce uint64, gasPrice *big.Int) *types.Transaction {
+	if nonce == 0 {
+		head := env.chain.CurrentHeader()
+		state, _ := env.chain.StateAt(head.Root)
+		nonce = state.GetNonce(address)
+	}
+	if gasPrice == nil {
+		gasPrice = big.NewInt(params.GWei)
+	}
+	tx, _ := types.SignTx(types.NewTransaction(nonce, common.Address{0x00}, big.NewInt(1000), params.TxGas, gasPrice, nil), signer, key)
+	return tx
+}
+
+func (env *testEnv) commit() {
+	head := env.chain.CurrentBlock()
+	block := env.chain.GetBlock(head.Hash(), head.Number.Uint64())
+	blocks, _ := core.GenerateChain(env.chain.Config(), block, ethash.NewFaker(), env.genDb, 1, func(i int, gen *core.BlockGen) {
+		tx, err := types.SignTx(types.NewTransaction(gen.TxNonce(address), common.Address{0x00}, big.NewInt(1000), params.TxGas, gen.BaseFee(), nil), signer, key)
+		if err != nil {
+			panic(err)
+		}
+		gen.AddTx(tx)
+	})
+	env.chain.InsertChain(blocks)
+	if err := env.pool.Sync(); err != nil {
+		panic(err)
+	}
+}
+
+func TestRejectInvalids(t *testing.T) {
+	env := newTestEnv(t, 10, 0, "")
+	defer env.close()
+
+	var cases = []struct {
+		gasTip uint64
+		tx     *types.Transaction
+		expErr error
+		commit bool
+	}{
+		{
+			tx:     env.makeTx(5, nil), // stale
+			expErr: core.ErrNonceTooLow,
+		},
+		{
+			tx:     env.makeTx(11, nil), // future transaction
+			expErr: nil,
+		},
+		{
+			gasTip: params.GWei,
+			tx:     env.makeTx(0, new(big.Int).SetUint64(params.GWei/2)), // low price
+			expErr: txpool.ErrUnderpriced,
+		},
+		{
+			tx:     types.NewTransaction(10, common.Address{0x00}, big.NewInt(1000), params.TxGas, big.NewInt(params.GWei), nil), // invalid signature
+			expErr: types.ErrInvalidSig,
+		},
+		{
+			commit: true,
+			tx:     env.makeTx(10, nil), // stale
+			expErr: core.ErrNonceTooLow,
+		},
+		{
+			tx:     env.makeTx(11, nil),
+			expErr: nil,
+		},
+	}
+	for i, c := range cases {
+		if c.gasTip != 0 {
+			env.setGasTip(c.gasTip)
+		}
+		if c.commit {
+			env.commit()
+		}
+		gotErr := env.tracker.Track(c.tx)
+		if c.expErr == nil && gotErr != nil {
+			t.Fatalf("%d, unexpected error: %v", i, gotErr)
+		}
+		if c.expErr != nil && !errors.Is(gotErr, c.expErr) {
+			t.Fatalf("%d, unexpected error, want: %v, got: %v", i, c.expErr, gotErr)
+		}
+	}
+}

core/txpool/txpool.go

@@ -24,11 +24,13 @@ import (
 
 	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/core"
+	"github.com/ethereum/go-ethereum/core/state"
 	"github.com/ethereum/go-ethereum/core/types"
 	"github.com/ethereum/go-ethereum/crypto/kzg4844"
 	"github.com/ethereum/go-ethereum/event"
 	"github.com/ethereum/go-ethereum/log"
 	"github.com/ethereum/go-ethereum/metrics"
+	"github.com/ethereum/go-ethereum/params"
 )
 
 // TxStatus is the current status of a transaction as seen by the pool.
@@ -53,11 +55,17 @@ var (
 // BlockChain defines the minimal set of methods needed to back a tx pool with
 // a chain. Exists to allow mocking the live chain out of tests.
 type BlockChain interface {
+	// Config retrieves the chain's fork configuration.
+	Config() *params.ChainConfig
+
 	// CurrentBlock returns the current head of the chain.
 	CurrentBlock() *types.Header
 
 	// SubscribeChainHeadEvent subscribes to new blocks being added to the chain.
 	SubscribeChainHeadEvent(ch chan<- core.ChainHeadEvent) event.Subscription
+
+	// StateAt returns a state database for a given root hash (generally the head).
+	StateAt(root common.Hash) (*state.StateDB, error)
 }
 
 // TxPool is an aggregator for various transaction specific pools, collectively
@@ -67,6 +75,11 @@ type BlockChain interface {
 // resource constraints.
 type TxPool struct {
 	subpools []SubPool // List of subpools for specialized transaction handling
+	chain    BlockChain
+	signer   types.Signer
+
+	stateLock sync.RWMutex   // The lock for protecting state instance
+	state     *state.StateDB // Current state at the blockchain head
 
 	reservations map[common.Address]SubPool // Map with the account to pool reservations
 	reserveLock  sync.Mutex                 // Lock protecting the account reservations
@@ -86,8 +99,21 @@ func New(gasTip uint64, chain BlockChain, subpools []SubPool) (*TxPool, error) {
 	// during initialization.
 	head := chain.CurrentBlock()
 
+	// Initialize the state with head block, or fallback to empty one in
+	// case the head state is not available (might occur when node is not
+	// fully synced).
+	statedb, err := chain.StateAt(head.Root)
+	if err != nil {
+		statedb, err = chain.StateAt(types.EmptyRootHash)
+	}
+	if err != nil {
+		return nil, err
+	}
 	pool := &TxPool{
 		subpools:     subpools,
+		chain:        chain,
+		signer:       types.LatestSigner(chain.Config()),
+		state:        statedb,
 		reservations: make(map[common.Address]SubPool),
 		quit:         make(chan chan error),
 		term:         make(chan struct{}),
@@ -101,7 +127,7 @@ func New(gasTip uint64, chain BlockChain, subpools []SubPool) (*TxPool, error) {
 			return nil, err
 		}
 	}
-	go pool.loop(head, chain)
+	go pool.loop(head)
 	return pool, nil
 }
 
@@ -179,14 +205,14 @@ func (p *TxPool) Close() error {
 // loop is the transaction pool's main event loop, waiting for and reacting to
 // outside blockchain events as well as for various reporting and transaction
 // eviction events.
-func (p *TxPool) loop(head *types.Header, chain BlockChain) {
+func (p *TxPool) loop(head *types.Header) {
 	// Close the termination marker when the pool stops
 	defer close(p.term)
 
 	// Subscribe to chain head events to trigger subpool resets
 	var (
 		newHeadCh  = make(chan core.ChainHeadEvent)
-		newHeadSub = chain.SubscribeChainHeadEvent(newHeadCh)
+		newHeadSub = p.chain.SubscribeChainHeadEvent(newHeadCh)
 	)
 	defer newHeadSub.Unsubscribe()
 
@@ -219,6 +245,14 @@ func (p *TxPool) loop(head *types.Header, chain BlockChain) {
 			// Try to inject a busy marker and start a reset if successful
 			select {
 			case resetBusy <- struct{}{}:
+				statedb, err := p.chain.StateAt(newHead.Root)
+				if err != nil {
+					log.Crit("Failed to reset txpool state", "err", err)
+				}
+				p.stateLock.Lock()
+				p.state = statedb
+				p.stateLock.Unlock()
+
 				// Busy marker injected, start a new subpool reset
 				go func(oldHead, newHead *types.Header) {
 					for _, subpool := range p.subpools {
@@ -339,6 +373,20 @@ func (p *TxPool) GetBlobs(vhashes []common.Hash) ([]*kzg4844.Blob, []*kzg4844.Pr
 // ValidateTxBasics checks whether a transaction is valid according to the consensus
 // rules, but does not check state-dependent validation such as sufficient balance.
 func (p *TxPool) ValidateTxBasics(tx *types.Transaction) error {
+	addr, err := types.Sender(p.signer, tx)
+	if err != nil {
+		return err
+	}
+	// Reject transactions with stale nonce. Gapped-nonce future transactions
+	// are considered valid and will be handled by the subpool according to its
+	// internal policy.
+	p.stateLock.RLock()
+	nonce := p.state.GetNonce(addr)
+	p.stateLock.RUnlock()
+
+	if nonce > tx.Nonce() {
+		return core.ErrNonceTooLow
+	}
 	for _, subpool := range p.subpools {
 		if subpool.Filter(tx) {
 			return subpool.ValidateTxBasics(tx)