forks/go-ethereum
1
0
Fork 1
mirror of https://github.com/ethereum/go-ethereum.git synced 2026-06-19 21:31:37 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 1

core/txpool: add 7702 protection to blobpool #31526 (#1930)

Browse source
This commit is contained in:
Daniel Liu 2026-01-26 13:50:02 +08:00 committed by GitHub
parent 48b6063261
commit a1b0d5476d
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
8 changed files with 205 additions and 151 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

4
core/txpool/errors.go
View file

@ -61,6 +61,10 @@ var (
// remains pending (and vice-versa).
ErrAlreadyReserved = errors.New("address already reserved")
// ErrInflightTxLimitReached is returned when the maximum number of in-flight
// transactions is reached for specific accounts.
ErrInflightTxLimitReached = errors.New("in-flight transaction limit reached for delegated accounts")
ErrZeroGasPrice = errors.New("zero gas price")
ErrUnderMinGasPrice = errors.New("under min gas price")

52
core/txpool/legacypool/legacypool.go
View file

@ -72,10 +72,6 @@ var (
// nonce received from the accounts with delegation or pending delegation.
ErrOutOfOrderTxFromDelegated = errors.New("gapped-nonce tx from delegated accounts")
// ErrInflightTxLimitReached is returned when the maximum number of in-flight
// transactions is reached for specific accounts.
ErrInflightTxLimitReached = errors.New("in-flight transaction limit reached for 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.
@ -259,11 +255,11 @@ type LegacyPool struct {
currentHead atomic.Pointer[types.Header] // Current head of the blockchain
currentState *state.StateDB // Current state in the blockchain head
pendingNonces *noncer // Pending state tracking virtual nonces
reserver *txpool.Reserver // Address reserver to ensure exclusivity across subpools
locals *accountSet // Set of local transaction to exempt from eviction rules
journal *journal // Journal of local transaction to back up to disk
reserve txpool.AddressReserver // Address reserver to ensure exclusivity across subpools
pending map[common.Address]*list // All currently processable transactions
queue map[common.Address]*list // Queued but non-processable transactions
beats map[common.Address]time.Time // Last heartbeat from each known account
@ -340,9 +336,9 @@ func (pool *LegacyPool) Filter(tx *types.Transaction) bool {
// head to allow balance / nonce checks. The transaction journal will be loaded
// from disk and filtered based on the provided starting settings. The internal
// goroutines will be spun up and the pool deemed operational afterwards.
func (pool *LegacyPool) Init(gasTip *big.Int, head *types.Header, reserve txpool.AddressReserver) error {
func (pool *LegacyPool) Init(gasTip *big.Int, head *types.Header, reserver *txpool.Reserver) error {
// Set the address reserver to request exclusive access to pooled accounts
pool.reserve = reserve
pool.reserver = reserver
// Set the basic pool parameters
pool.gasTip.Store(gasTip)
@ -750,7 +746,7 @@ func (pool *LegacyPool) checkDelegationLimit(tx *types.Transaction) error {
if pending.Contains(tx.Nonce()) {
return nil
}
return ErrInflightTxLimitReached
return txpool.ErrInflightTxLimitReached
}
// validateAuth verifies that the transaction complies with code authorization
@ -761,12 +757,24 @@ func (pool *LegacyPool) validateAuth(tx *types.Transaction) error {
if err := pool.checkDelegationLimit(tx); err != nil {
return err
}
// Authorities cannot conflict with any pending or queued transactions.
// Authorities must not conflict with any pending or queued transactions,
// nor with addresses that have already been reserved.
if auths := tx.SetCodeAuthorities(); len(auths) > 0 {
for _, auth := range auths {
if pool.pending[auth] != nil || pool.queue[auth] != nil {
return ErrAuthorityReserved
}
// Because there is no exclusive lock held between different subpools
// when processing transactions, the SetCode transaction may be accepted
// while other transactions with the same sender address are also
// accepted simultaneously in the other pools.
//
// This scenario is considered acceptable, as the rule primarily ensures
// that attackers cannot easily stack a SetCode transaction when the sender
// is reserved by other pools.
if pool.reserver.Has(auth) {
return ErrAuthorityReserved
}
}
}
return nil
@ -811,7 +819,7 @@ func (pool *LegacyPool) add(tx *types.Transaction, local bool) (replaced bool, e
_, hasQueued = pool.queue[from]
)
if !hasPending && !hasQueued {
if err := pool.reserve(from, true); err != nil {
if err := pool.reserver.Hold(from); err != nil {
return false, err
}
defer func() {
@ -822,7 +830,7 @@ func (pool *LegacyPool) add(tx *types.Transaction, local bool) (replaced bool, e
// by a return statement before running deferred methods. Take care with
// removing or subscoping err as it will break this clause.
if err != nil {
pool.reserve(from, false)
pool.reserver.Release(from)
}
}()
}
@ -1270,7 +1278,7 @@ func (pool *LegacyPool) removeTx(hash common.Hash, outofbound bool, unreserve bo
_, hasQueued = pool.queue[addr]
)
if !hasPending && !hasQueued {
pool.reserve(addr, false)
pool.reserver.Release(addr)
}
}()
}
@ -1666,7 +1674,7 @@ func (pool *LegacyPool) promoteExecutables(accounts []common.Address) []*types.T
delete(pool.queue, addr)
delete(pool.beats, addr)
if _, ok := pool.pending[addr]; !ok {
pool.reserve(addr, false)
pool.reserver.Release(addr)
}
}
}
@ -1866,7 +1874,7 @@ func (pool *LegacyPool) demoteUnexecutables() {
if list.Empty() {
delete(pool.pending, addr)
if _, ok := pool.queue[addr]; !ok {
pool.reserve(addr, false)
pool.reserver.Release(addr)
}
}
}
@ -2203,8 +2211,8 @@ func (pool *LegacyPool) Clear() {
pool.mu.Lock()
defer pool.mu.Unlock()
// unreserve each tracked account. Ideally, we could just clear the
// reservation map in the parent txpool context. However, if we clear in
// unreserve each tracked account. Ideally, we could just clear the
// reservation map in the parent txpool context. However, if we clear in
// parent context, to avoid exposing the subpool lock, we have to lock the
// reservations and then lock each subpool.
//
@ -2215,15 +2223,15 @@ func (pool *LegacyPool) Clear() {
// * TxPool.Clear attempts to lock subpool mutex
//
// The transaction addition may attempt to reserve the sender addr which
// can't happen until Clear releases the reservation lock. Clear cannot
// can't happen until Clear releases the reservation lock. Clear cannot
// acquire the subpool lock until the transaction addition is completed.
for _, tx := range pool.all.locals {
senderAddr, _ := types.Sender(pool.signer, tx)
pool.reserve(senderAddr, false)
pool.reserver.Release(senderAddr)
}
for _, tx := range pool.all.remotes {
senderAddr, _ := types.Sender(pool.signer, tx)
pool.reserve(senderAddr, false)
pool.reserver.Release(senderAddr)
}
pool.all = newLookup()
pool.priced = newPricedList(pool.all)
@ -2231,3 +2239,9 @@ func (pool *LegacyPool) Clear() {
pool.queue = make(map[common.Address]*list)
pool.pendingNonces = newNoncer(pool.currentState)
}
// HasPendingAuth returns a flag indicating whether there are pending
// authorizations from the specific address cached in the pool.
func (pool *LegacyPool) HasPendingAuth(addr common.Address) bool {
return pool.all.hasAuth(addr)
}

8
core/txpool/legacypool/legacypool2_test.go
View file

@ -85,7 +85,7 @@ func TestTransactionFutureAttack(t *testing.T) {
config.GlobalQueue = 100
config.GlobalSlots = 100
pool := New(config, blockchain)
pool.Init(new(big.Int).SetUint64(config.PriceLimit), blockchain.CurrentBlock(), makeAddressReserver())
pool.Init(new(big.Int).SetUint64(config.PriceLimit), blockchain.CurrentBlock(), newReserver())
defer pool.Close()
fillPool(t, pool)
pending, _ := pool.Stats()
@ -119,7 +119,7 @@ func TestTransactionFuture1559(t *testing.T) {
statedb, _ := state.New(types.EmptyRootHash, state.NewDatabase(rawdb.NewMemoryDatabase()))
blockchain := newTestBlockChain(eip1559Config, 1000000, statedb, new(event.Feed))
pool := New(testTxPoolConfig, blockchain)
pool.Init(new(big.Int).SetUint64(testTxPoolConfig.PriceLimit), blockchain.CurrentBlock(), makeAddressReserver())
pool.Init(new(big.Int).SetUint64(testTxPoolConfig.PriceLimit), blockchain.CurrentBlock(), newReserver())
defer pool.Close()
// Create a number of test accounts, fund them and make transactions
@ -152,7 +152,7 @@ func TestTransactionZAttack(t *testing.T) {
statedb, _ := state.New(types.EmptyRootHash, state.NewDatabase(rawdb.NewMemoryDatabase()))
blockchain := newTestBlockChain(eip1559Config, 1000000, statedb, new(event.Feed))
pool := New(testTxPoolConfig, blockchain)
pool.Init(new(big.Int).SetUint64(testTxPoolConfig.PriceLimit), blockchain.CurrentBlock(), makeAddressReserver())
pool.Init(new(big.Int).SetUint64(testTxPoolConfig.PriceLimit), blockchain.CurrentBlock(), newReserver())
defer pool.Close()
// Create a number of test accounts, fund them and make transactions
fillPool(t, pool)
@ -226,7 +226,7 @@ func BenchmarkFutureAttack(b *testing.B) {
config.GlobalQueue = 100
config.GlobalSlots = 100
pool := New(config, blockchain)
pool.Init(new(big.Int).SetUint64(testTxPoolConfig.PriceLimit), blockchain.CurrentBlock(), makeAddressReserver())
pool.Init(new(big.Int).SetUint64(testTxPoolConfig.PriceLimit), blockchain.CurrentBlock(), newReserver())
defer pool.Close()
fillPool(b, pool)

90
core/txpool/legacypool/legacypool_test.go
View file

@ -25,7 +25,6 @@ import (
"math/big"
"math/rand"
"os"
"sync"
"sync/atomic"
"testing"
"time"
@ -184,35 +183,14 @@ func pricedSetCodeTxWithAuth(nonce uint64, gaslimit uint64, gasFee, tip *uint256
})
}
func makeAddressReserver() txpool.AddressReserver {
var (
reserved = make(map[common.Address]struct{})
lock sync.Mutex
)
return func(addr common.Address, reserve bool) error {
lock.Lock()
defer lock.Unlock()
_, exists := reserved[addr]
if reserve {
if exists {
panic("already reserved")
}
reserved[addr] = struct{}{}
return nil
}
if !exists {
panic("not reserved")
}
delete(reserved, addr)
return nil
}
}
func setupPool() (*LegacyPool, *ecdsa.PrivateKey) {
return setupPoolWithConfig(params.TestChainConfig)
}
func newReserver() *txpool.Reserver {
return txpool.NewReservationTracker().NewHandle(42)
}
func setupPoolWithConfig(config *params.ChainConfig) (*LegacyPool, *ecdsa.PrivateKey) {
diskdb := rawdb.NewMemoryDatabase()
statedb, _ := state.New(types.EmptyRootHash, state.NewDatabase(diskdb))
@ -220,7 +198,7 @@ func setupPoolWithConfig(config *params.ChainConfig) (*LegacyPool, *ecdsa.Privat
key, _ := crypto.GenerateKey()
pool := New(testTxPoolConfig, blockchain)
if err := pool.Init(new(big.Int).SetUint64(testTxPoolConfig.PriceLimit), blockchain.CurrentBlock(), makeAddressReserver()); err != nil {
if err := pool.Init(new(big.Int).SetUint64(testTxPoolConfig.PriceLimit), blockchain.CurrentBlock(), newReserver()); err != nil {
panic(err)
}
// wait for the pool to initialize
@ -340,7 +318,7 @@ func TestStateChangeDuringReset(t *testing.T) {
tx1 := pricedTransaction(1, 100000, big.NewInt(250000000), key)
pool := New(testTxPoolConfig, blockchain)
pool.Init(new(big.Int).SetUint64(testTxPoolConfig.PriceLimit), blockchain.CurrentBlock(), makeAddressReserver())
pool.Init(new(big.Int).SetUint64(testTxPoolConfig.PriceLimit), blockchain.CurrentBlock(), newReserver())
defer pool.Close()
nonce := pool.Nonce(address)
@ -835,7 +813,7 @@ func TestPostponing(t *testing.T) {
blockchain := newTestBlockChain(params.TestChainConfig, 1000000, statedb, new(event.Feed))
pool := New(testTxPoolConfig, blockchain)
pool.Init(new(big.Int).SetUint64(testTxPoolConfig.PriceLimit), blockchain.CurrentBlock(), makeAddressReserver())
pool.Init(new(big.Int).SetUint64(testTxPoolConfig.PriceLimit), blockchain.CurrentBlock(), newReserver())
defer pool.Close()
// Create two test accounts to produce different gap profiles with
@ -1053,7 +1031,7 @@ func testQueueGlobalLimiting(t *testing.T, nolocals bool) {
config.GlobalQueue = config.AccountQueue*3 - 1 // reduce the queue limits to shorten test time (-1 to make it non divisible)
pool := New(config, blockchain)
pool.Init(new(big.Int).SetUint64(testTxPoolConfig.PriceLimit), blockchain.CurrentBlock(), makeAddressReserver())
pool.Init(new(big.Int).SetUint64(testTxPoolConfig.PriceLimit), blockchain.CurrentBlock(), newReserver())
defer pool.Close()
// Create a number of test accounts and fund them (last one will be the local)
@ -1144,7 +1122,7 @@ func testQueueTimeLimiting(t *testing.T, nolocals bool) {
config.NoLocals = nolocals
pool := New(config, blockchain)
pool.Init(new(big.Int).SetUint64(config.PriceLimit), blockchain.CurrentBlock(), makeAddressReserver())
pool.Init(new(big.Int).SetUint64(config.PriceLimit), blockchain.CurrentBlock(), newReserver())
defer pool.Close()
// Create two test accounts to ensure remotes expire but locals do not
@ -1330,7 +1308,7 @@ func TestPendingGlobalLimiting(t *testing.T) {
config.GlobalSlots = config.AccountSlots * 10
pool := New(config, blockchain)
pool.Init(new(big.Int).SetUint64(config.PriceLimit), blockchain.CurrentBlock(), makeAddressReserver())
pool.Init(new(big.Int).SetUint64(config.PriceLimit), blockchain.CurrentBlock(), newReserver())
defer pool.Close()
// Create a number of test accounts and fund them
@ -1434,7 +1412,7 @@ func TestCapClearsFromAll(t *testing.T) {
config.GlobalSlots = 8
pool := New(config, blockchain)
pool.Init(new(big.Int).SetUint64(config.PriceLimit), blockchain.CurrentBlock(), makeAddressReserver())
pool.Init(new(big.Int).SetUint64(config.PriceLimit), blockchain.CurrentBlock(), newReserver())
defer pool.Close()
// Create a number of test accounts and fund them
@ -1468,7 +1446,7 @@ func TestPendingMinimumAllowance(t *testing.T) {
config.GlobalSlots = 1
pool := New(config, blockchain)
pool.Init(new(big.Int).SetUint64(config.PriceLimit), blockchain.CurrentBlock(), makeAddressReserver())
pool.Init(new(big.Int).SetUint64(config.PriceLimit), blockchain.CurrentBlock(), newReserver())
defer pool.Close()
// Create a number of test accounts and fund them
@ -1514,7 +1492,7 @@ func TestRepricing(t *testing.T) {
blockchain := newTestBlockChain(params.TestChainConfig, 1000000, statedb, new(event.Feed))
pool := New(testTxPoolConfig, blockchain)
pool.Init(new(big.Int).SetUint64(testTxPoolConfig.PriceLimit), blockchain.CurrentBlock(), makeAddressReserver())
pool.Init(new(big.Int).SetUint64(testTxPoolConfig.PriceLimit), blockchain.CurrentBlock(), newReserver())
defer pool.Close()
// Keep track of transaction events to ensure all executables get announced
@ -1763,7 +1741,7 @@ func TestRepricingKeepsLocals(t *testing.T) {
blockchain := newTestBlockChain(eip1559Config, 1000000, statedb, new(event.Feed))
pool := New(testTxPoolConfig, blockchain)
pool.Init(new(big.Int).SetUint64(testTxPoolConfig.PriceLimit), blockchain.CurrentBlock(), makeAddressReserver())
pool.Init(new(big.Int).SetUint64(testTxPoolConfig.PriceLimit), blockchain.CurrentBlock(), newReserver())
defer pool.Close()
// Create a number of test accounts and fund them
@ -1842,7 +1820,7 @@ func TestUnderpricing(t *testing.T) {
config.GlobalQueue = 2
pool := New(config, blockchain)
pool.Init(new(big.Int).SetUint64(config.PriceLimit), blockchain.CurrentBlock(), makeAddressReserver())
pool.Init(new(big.Int).SetUint64(config.PriceLimit), blockchain.CurrentBlock(), newReserver())
defer pool.Close()
// Keep track of transaction events to ensure all executables get announced
@ -1961,7 +1939,7 @@ func TestStableUnderpricing(t *testing.T) {
config.AccountSlots = config.GlobalSlots - 1
pool := New(config, blockchain)
pool.Init(new(big.Int).SetUint64(config.PriceLimit), blockchain.CurrentBlock(), makeAddressReserver())
pool.Init(new(big.Int).SetUint64(config.PriceLimit), blockchain.CurrentBlock(), newReserver())
defer pool.Close()
// Keep track of transaction events to ensure all executables get announced
@ -2190,7 +2168,7 @@ func TestDeduplication(t *testing.T) {
blockchain := newTestBlockChain(params.TestChainConfig, 1000000, statedb, new(event.Feed))
pool := New(testTxPoolConfig, blockchain)
pool.Init(new(big.Int).SetUint64(testTxPoolConfig.PriceLimit), blockchain.CurrentBlock(), makeAddressReserver())
pool.Init(new(big.Int).SetUint64(testTxPoolConfig.PriceLimit), blockchain.CurrentBlock(), newReserver())
defer pool.Close()
// Create a test account to add transactions with
@ -2257,7 +2235,7 @@ func TestReplacement(t *testing.T) {
blockchain := newTestBlockChain(params.TestChainConfig, 1000000, statedb, new(event.Feed))
pool := New(testTxPoolConfig, blockchain)
pool.Init(new(big.Int).SetUint64(testTxPoolConfig.PriceLimit), blockchain.CurrentBlock(), makeAddressReserver())
pool.Init(new(big.Int).SetUint64(testTxPoolConfig.PriceLimit), blockchain.CurrentBlock(), newReserver())
defer pool.Close()
// Keep track of transaction events to ensure all executables get announced
@ -2469,7 +2447,7 @@ func testJournaling(t *testing.T, nolocals bool) {
config.Rejournal = time.Second
pool := New(config, blockchain)
pool.Init(new(big.Int).SetUint64(config.PriceLimit), blockchain.CurrentBlock(), makeAddressReserver())
pool.Init(new(big.Int).SetUint64(config.PriceLimit), blockchain.CurrentBlock(), newReserver())
// Create two test accounts to ensure remotes expire but locals do not
local, _ := crypto.GenerateKey()
@ -2507,7 +2485,7 @@ func testJournaling(t *testing.T, nolocals bool) {
blockchain = newTestBlockChain(params.TestChainConfig, 1000000, statedb, new(event.Feed))
pool = New(config, blockchain)
pool.Init(new(big.Int).SetUint64(config.PriceLimit), blockchain.CurrentBlock(), makeAddressReserver())
pool.Init(new(big.Int).SetUint64(config.PriceLimit), blockchain.CurrentBlock(), newReserver())
pending, queued = pool.Stats()
if queued != 0 {
@ -2534,7 +2512,7 @@ func testJournaling(t *testing.T, nolocals bool) {
statedb.SetNonce(crypto.PubkeyToAddress(local.PublicKey), 1)
blockchain = newTestBlockChain(params.TestChainConfig, 1000000, statedb, new(event.Feed))
pool = New(config, blockchain)
pool.Init(new(big.Int).SetUint64(config.PriceLimit), blockchain.CurrentBlock(), makeAddressReserver())
pool.Init(new(big.Int).SetUint64(config.PriceLimit), blockchain.CurrentBlock(), newReserver())
pending, queued = pool.Stats()
if pending != 0 {
@ -2565,7 +2543,7 @@ func TestStatusCheck(t *testing.T) {
blockchain := newTestBlockChain(params.TestChainConfig, 1000000, statedb, new(event.Feed))
pool := New(testTxPoolConfig, blockchain)
pool.Init(new(big.Int).SetUint64(testTxPoolConfig.PriceLimit), blockchain.CurrentBlock(), makeAddressReserver())
pool.Init(new(big.Int).SetUint64(testTxPoolConfig.PriceLimit), blockchain.CurrentBlock(), newReserver())
defer pool.Close()
// Create the test accounts to check various transaction statuses with
@ -2638,7 +2616,7 @@ func TestSetCodeTransactions(t *testing.T) {
blockchain := newTestBlockChain(params.MergedTestChainConfig, 1000000, statedb, new(event.Feed))
pool := New(testTxPoolConfig, blockchain)
pool.Init(new(big.Int).SetUint64(testTxPoolConfig.PriceLimit), blockchain.CurrentBlock(), makeAddressReserver())
pool.Init(new(big.Int).SetUint64(testTxPoolConfig.PriceLimit), blockchain.CurrentBlock(), newReserver())
defer pool.Close()
// Create the test accounts
@ -2685,12 +2663,12 @@ func TestSetCodeTransactions(t *testing.T) {
if err := pool.addRemoteSync(pricedTransaction(0, 100000, minGasPrice, keyA)); err != nil {
t.Fatalf("%s: failed to add remote transaction: %v", name, err)
}
if err := pool.addRemoteSync(pricedTransaction(1, 100000, minGasPrice, keyA)); !errors.Is(err, ErrInflightTxLimitReached) {
t.Fatalf("%s: error mismatch: want %v, have %v", name, ErrInflightTxLimitReached, err)
if err := pool.addRemoteSync(pricedTransaction(1, 100000, minGasPrice, keyA)); !errors.Is(err, txpool.ErrInflightTxLimitReached) {
t.Fatalf("%s: error mismatch: want %v, have %v", name, txpool.ErrInflightTxLimitReached, err)
}
// Check gapped transaction again.
if err := pool.addRemoteSync(pricedTransaction(2, 100000, minGasPrice, keyA)); !errors.Is(err, ErrInflightTxLimitReached) {
t.Fatalf("%s: error mismatch: want %v, have %v", name, ErrInflightTxLimitReached, err)
if err := pool.addRemoteSync(pricedTransaction(2, 100000, minGasPrice, keyA)); !errors.Is(err, txpool.ErrInflightTxLimitReached) {
t.Fatalf("%s: error mismatch: want %v, have %v", name, txpool.ErrInflightTxLimitReached, err)
}
// Replace by fee.
if err := pool.addRemoteSync(pricedTransaction(0, 100000, legacyReplacePrice, keyA)); err != nil {
@ -2721,11 +2699,11 @@ func TestSetCodeTransactions(t *testing.T) {
t.Fatalf("%s: failed to add with remote setcode transaction: %v", name, err)
}
if err := pool.addRemoteSync(pricedTransaction(0, 100000, minGasPrice, keyC)); err != nil {
t.Fatalf("%s: failed to add with pending delegatio: %v", name, err)
t.Fatalf("%s: failed to add with pending delegation: %v", name, err)
}
// Also check gapped transaction is rejected.
if err := pool.addRemoteSync(pricedTransaction(1, 100000, minGasPrice, keyC)); !errors.Is(err, ErrInflightTxLimitReached) {
t.Fatalf("%s: error mismatch: want %v, have %v", name, ErrInflightTxLimitReached, err)
if err := pool.addRemoteSync(pricedTransaction(1, 100000, minGasPrice, keyC)); !errors.Is(err, txpool.ErrInflightTxLimitReached) {
t.Fatalf("%s: error mismatch: want %v, have %v", name, txpool.ErrInflightTxLimitReached, err)
}
},
},
@ -2800,7 +2778,7 @@ func TestSetCodeTransactions(t *testing.T) {
if err := pool.addRemoteSync(pricedTransaction(0, 100000, minGasPrice, keyC)); err != nil {
t.Fatalf("%s: failed to added single pooled for account with pending delegation: %v", name, err)
}
if err, want := pool.addRemoteSync(pricedTransaction(1, 100000, minGasPrice, keyC)), ErrInflightTxLimitReached; !errors.Is(err, want) {
if err, want := pool.addRemoteSync(pricedTransaction(1, 100000, minGasPrice, keyC)), txpool.ErrInflightTxLimitReached; !errors.Is(err, want) {
t.Fatalf("%s: error mismatch: want %v, have %v", name, want, err)
}
},
@ -2842,7 +2820,7 @@ func TestSetCodeTransactionsReorg(t *testing.T) {
blockchain := newTestBlockChain(params.MergedTestChainConfig, 1000000, statedb, new(event.Feed))
pool := New(testTxPoolConfig, blockchain)
pool.Init(new(big.Int).SetUint64(testTxPoolConfig.PriceLimit), blockchain.CurrentBlock(), makeAddressReserver())
pool.Init(new(big.Int).SetUint64(testTxPoolConfig.PriceLimit), blockchain.CurrentBlock(), newReserver())
defer pool.Close()
// Create the test accounts
@ -2882,8 +2860,8 @@ func TestSetCodeTransactionsReorg(t *testing.T) {
t.Fatalf("failed to add with remote setcode transaction: %v", err)
}
// Try to add a transactions in
if err := pool.addRemoteSync(pricedTransaction(2, 100000, new(big.Int).Set(legacyPrice), keyA)); !errors.Is(err, ErrInflightTxLimitReached) {
t.Fatalf("unexpected error %v, expecting %v", err, ErrInflightTxLimitReached)
if err := pool.addRemoteSync(pricedTransaction(2, 100000, new(big.Int).Set(legacyPrice), keyA)); !errors.Is(err, txpool.ErrInflightTxLimitReached) {
t.Fatalf("unexpected error %v, expecting %v", err, txpool.ErrInflightTxLimitReached)
}
// Simulate the chain moving
blockchain.statedb.SetNonce(addrA, 2)

124
core/txpool/reserver.go Normal file
View file

@ -0,0 +1,124 @@
// 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 txpool
import (
"errors"
"fmt"
"sync"
"github.com/XinFinOrg/XDPoSChain/common"
"github.com/XinFinOrg/XDPoSChain/log"
"github.com/XinFinOrg/XDPoSChain/metrics"
)
var (
// reservationsGaugeName is the prefix of a per-subpool address reservation
// metric.
//
// This is mostly a sanity metric to ensure there's no bug that would make
// some subpool hog all the reservations due to mis-accounting.
reservationsGaugeName = "txpool/reservations"
)
// ReservationTracker is a struct shared between different subpools. It is used to reserve
// the account and ensure that one address cannot initiate transactions, authorizations,
// and other state-changing behaviors in different pools at the same time.
type ReservationTracker struct {
accounts map[common.Address]int
lock sync.RWMutex
}
// NewReservationTracker initializes the account reservation tracker.
func NewReservationTracker() *ReservationTracker {
return &ReservationTracker{
accounts: make(map[common.Address]int),
}
}
// NewHandle creates a named handle on the ReservationTracker. The handle
// identifies the subpool so ownership of reservations can be determined.
func (r *ReservationTracker) NewHandle(id int) *Reserver {
return &Reserver{r, id}
}
// Reserver is a named handle on ReservationTracker. It is held by subpools to
// make reservations for accounts it is tracking. The id is used to determine
// which pool owns an address and disallows non-owners to hold or release
// addresses it doesn't own.
type Reserver struct {
tracker *ReservationTracker
id int
}
// Hold attempts to reserve the specified account address for the given pool.
// Returns an error if the account is already reserved.
func (h *Reserver) Hold(addr common.Address) error {
h.tracker.lock.Lock()
defer h.tracker.lock.Unlock()
// Double reservations are forbidden even from the same pool to
// avoid subtle bugs in the long term.
owner, exists := h.tracker.accounts[addr]
if exists {
if owner == h.id {
log.Error("pool attempted to reserve already-owned address", "address", addr)
return nil // Ignore fault to give the pool a chance to recover while the bug gets fixed
}
return ErrAlreadyReserved
}
h.tracker.accounts[addr] = h.id
if metrics.Enabled() {
m := fmt.Sprintf("%s/%d", reservationsGaugeName, h.id)
metrics.GetOrRegisterGauge(m, nil).Inc(1)
}
return nil
}
// Release attempts to release the reservation for the specified account.
// Returns an error if the address is not reserved or is reserved by another pool.
func (h *Reserver) Release(addr common.Address) error {
h.tracker.lock.Lock()
defer h.tracker.lock.Unlock()
// Ensure subpools only attempt to unreserve their own owned addresses,
// otherwise flag as a programming error.
owner, exists := h.tracker.accounts[addr]
if !exists {
log.Error("pool attempted to unreserve non-reserved address", "address", addr)
return errors.New("address not reserved")
}
if owner != h.id {
log.Error("pool attempted to unreserve non-owned address", "address", addr)
return errors.New("address not owned")
}
delete(h.tracker.accounts, addr)
if metrics.Enabled() {
m := fmt.Sprintf("%s/%d", reservationsGaugeName, h.id)
metrics.GetOrRegisterGauge(m, nil).Dec(1)
}
return nil
}
// Has returns a flag indicating if the address has been reserved or not.
func (h *Reserver) Has(address common.Address) bool {
h.tracker.lock.RLock()
defer h.tracker.lock.RUnlock()
_, exists := h.tracker.accounts[address]
return exists
}

6
core/txpool/subpool.go
View file

@ -54,10 +54,6 @@ func (ltx *LazyTransaction) Resolve() *Transaction {
return ltx.Tx
}
// AddressReserver is passed by the main transaction pool to subpools, so they
// may request (and relinquish) exclusive access to certain addresses.
type AddressReserver func(addr common.Address, reserve bool) error
// SubPool represents a specialized transaction pool that lives on its own (e.g.
// blob pool). Since independent of how many specialized pools we have, they do
// need to be updated in lockstep and assemble into one coherent view for block
@ -75,7 +71,7 @@ type SubPool interface {
// These should not be passed as a constructor argument - nor should the pools
// start by themselves - in order to keep multiple subpools in lockstep with
// one another.
Init(gasTip *big.Int, head *types.Header, reserve AddressReserver) error
Init(gasTip *big.Int, head *types.Header, reserver *Reserver) error
// Close terminates any background processing threads and releases any held
// resources.

70
core/txpool/txpool.go
View file

@ -17,18 +17,14 @@
package txpool
import (
"errors"
"fmt"
"maps"
"math/big"
"sync"
"github.com/XinFinOrg/XDPoSChain/common"
"github.com/XinFinOrg/XDPoSChain/core"
"github.com/XinFinOrg/XDPoSChain/core/types"
"github.com/XinFinOrg/XDPoSChain/event"
"github.com/XinFinOrg/XDPoSChain/log"
"github.com/XinFinOrg/XDPoSChain/metrics"
)
// TxStatus is the current status of a transaction as seen by the pool.
@ -41,15 +37,6 @@ const (
TxStatusIncluded
)
var (
// reservationsGaugeName is the prefix of a per-subpool address reservation
// metric.
//
// This is mostly a sanity metric to ensure there's no bug that would make
// some subpool hog all the reservations due to mis-accounting.
reservationsGaugeName = "txpool/reservations"
)
// 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 {
@ -68,9 +55,6 @@ type BlockChain interface {
type TxPool struct {
subpools []SubPool // List of subpools for specialized transaction handling
reservations map[common.Address]SubPool // Map with the account to pool reservations
reserveLock sync.Mutex // Lock protecting the account reservations
subs event.SubscriptionScope // Subscription scope to unsubscribe all on shutdown
quit chan chan error // Quit channel to tear down the head updater
}
@ -84,12 +68,12 @@ func New(gasTip *big.Int, chain BlockChain, subpools []SubPool) (*TxPool, error)
head := chain.CurrentBlock()
pool := &TxPool{
subpools: subpools,
reservations: make(map[common.Address]SubPool),
quit: make(chan chan error),
subpools: subpools,
quit: make(chan chan error),
}
reserver := NewReservationTracker()
for i, subpool := range subpools {
if err := subpool.Init(gasTip, head, pool.reserver(i, subpool)); err != nil {
if err := subpool.Init(gasTip, head, reserver.NewHandle(i)); err != nil {
for j := i - 1; j >= 0; j-- {
subpools[j].Close()
}
@ -100,52 +84,6 @@ func New(gasTip *big.Int, chain BlockChain, subpools []SubPool) (*TxPool, error)
return pool, nil
}
// reserver is a method to create an address reservation callback to exclusively
// assign/deassign addresses to/from subpools. This can ensure that at any point
// in time, only a single subpool is able to manage an account, avoiding cross
// subpool eviction issues and nonce conflicts.
func (p *TxPool) reserver(id int, subpool SubPool) AddressReserver {
return func(addr common.Address, reserve bool) error {
p.reserveLock.Lock()
defer p.reserveLock.Unlock()
owner, exists := p.reservations[addr]
if reserve {
// Double reservations are forbidden even from the same pool to
// avoid subtle bugs in the long term.
if exists {
if owner == subpool {
log.Error("pool attempted to reserve already-owned address", "address", addr)
return nil // Ignore fault to give the pool a chance to recover while the bug gets fixed
}
return ErrAlreadyReserved
}
p.reservations[addr] = subpool
if metrics.Enabled() {
m := fmt.Sprintf("%s/%d", reservationsGaugeName, id)
metrics.GetOrRegisterGauge(m, nil).Inc(1)
}
return nil
}
// Ensure subpools only attempt to unreserve their own owned addresses,
// otherwise flag as a programming error.
if !exists {
log.Error("pool attempted to unreserve non-reserved address", "address", addr)
return errors.New("address not reserved")
}
if subpool != owner {
log.Error("pool attempted to unreserve non-owned address", "address", addr)
return errors.New("address not owned")
}
delete(p.reservations, addr)
if metrics.Enabled() {
m := fmt.Sprintf("%s/%d", reservationsGaugeName, id)
metrics.GetOrRegisterGauge(m, nil).Dec(1)
}
return nil
}
}
// Close terminates the transaction pool and all its subpools.
func (p *TxPool) Close() error {
var errs []error

2
core/txpool/validation.go
View file

@ -225,7 +225,7 @@ func ValidateTransactionWithState(tx *types.Transaction, signer types.Signer, op
return fmt.Errorf("%w: balance %v, queued cost %v, tx cost %v, overshot %v", core.ErrInsufficientFunds, balance, spent, cost, new(big.Int).Sub(need, newBalance))
}
// Transaction takes a new nonce value out of the pool. Ensure it doesn't
// overflow the number of permitted transactions from a single accoun
// overflow the number of permitted transactions from a single account
// (i.e. max cancellable via out-of-bound transaction).
if opts.UsedAndLeftSlots != nil {
if used, left := opts.UsedAndLeftSlots(from); left <= 0 {