forks/go-ethereum
1
0
Fork 1
mirror of https://github.com/ethereum/go-ethereum.git synced 2026-07-27 15:16:43 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 3

core/txpool: add 7702 protection to blobpool

Browse source
This commit is contained in:
Gary Rong 2025-03-28 15:20:48 +08:00
parent 77dc1acafa
commit 6fd341371b
10 changed files with 207 additions and 86 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

41
core/txpool/blobpool/blobpool.go
View file

@ -300,6 +300,7 @@ func newBlobTxMeta(id uint64, size uint64, storageSize uint32, tx *types.Transac
type BlobPool struct {
config Config // Pool configuration
reserve txpool.AddressReserver // Address reserver to ensure exclusivity across subpools
hasPendingAuth func(common.Address) bool // Determine whether the specified address has a pending 7702-auth
store billy.Database // Persistent data store for the tx metadata and blobs
stored uint64 // Useful data size of all transactions on disk
@ -329,13 +330,14 @@ type BlobPool struct {
// New creates a new blob transaction pool to gather, sort and filter inbound
// blob transactions from the network.
func New(config Config, chain BlockChain) *BlobPool {
func New(config Config, chain BlockChain, hasPendingAuth func(common.Address) bool) *BlobPool {
// Sanitize the input to ensure no vulnerable gas prices are set
config = (&config).sanitize()
// Create the transaction pool with its initial settings
return &BlobPool{
config: config,
hasPendingAuth: hasPendingAuth,
signer: types.LatestSigner(chain.Config()),
chain: chain,
lookup: newLookup(),
@ -353,7 +355,7 @@ func (p *BlobPool) Filter(tx *types.Transaction) bool {
// Init sets the gas price needed to keep a transaction in the pool and the chain
// head to allow balance / nonce checks. The transaction journal will be loaded
// from disk and filtered based on the provided starting settings.
func (p *BlobPool) Init(gasTip uint64, head *types.Header, reserve txpool.AddressReserver) error {
func (p *BlobPool) Init(gasTip uint64, head *types.Header, reserve txpool.AddressReserver, _ txpool.IsAddressReserved) error {
p.reserve = reserve
var (
@ -1101,6 +1103,38 @@ func (p *BlobPool) ValidateTxBasics(tx *types.Transaction) error {
return txpool.ValidateTransaction(tx, p.head, p.signer, opts)
}
// checkDelegationLimit determines if the tx sender is delegated or has a
// pending delegation, and if so, ensures they have at most one in-flight
// **executable** transaction, e.g. disallow stacked and gapped transactions
// from the account.
func (p *BlobPool) checkDelegationLimit(tx *types.Transaction) error {
from, _ := types.Sender(p.signer, tx) // validated
// Short circuit if the sender has neither delegation nor pending delegation.
if p.state.GetCodeHash(from) == types.EmptyCodeHash {
// Because there is no exclusive lock held between different subpools
// when processing transactions, a blob transaction may be accepted
// while other SetCode transactions with pending authorities from the
// same address are also accepted simultaneously.
//
// This scenario is considered acceptable, as the rule primarily ensures
// that attackers cannot easily and endlessly stack blob transactions
// with a delegated or pending delegated sender.
if p.hasPendingAuth == nil || !p.hasPendingAuth(from) {
return nil
}
}
pending := p.index[from]
if len(pending) == 0 {
return nil
}
// Transaction replacement is supported
if len(pending) == 1 && pending[0].nonce == tx.Nonce() {
return nil
}
return txpool.ErrInflightTxLimitReached
}
// validateTx checks whether a transaction is valid according to the consensus
// rules and adheres to some heuristic limits of the local node (price and size).
func (p *BlobPool) validateTx(tx *types.Transaction) error {
@ -1141,6 +1175,9 @@ func (p *BlobPool) validateTx(tx *types.Transaction) error {
if err := txpool.ValidateTransactionWithState(tx, p.signer, stateOpts); err != nil {
return err
}
if err := p.checkDelegationLimit(tx); err != nil {
return err
}
// If the transaction replaces an existing one, ensure that price bumps are
// adhered to.
var (

28
core/txpool/blobpool/blobpool_test.go
View file

@ -699,8 +699,8 @@ func TestOpenDrops(t *testing.T) {
blobfee: uint256.NewInt(params.BlobTxMinBlobGasprice),
statedb: statedb,
}
pool := New(Config{Datadir: storage}, chain)
if err := pool.Init(1, chain.CurrentBlock(), makeAddressReserver()); err != nil {
pool := New(Config{Datadir: storage}, chain, nil)
if err := pool.Init(1, chain.CurrentBlock(), makeAddressReserver(), nil); err != nil {
t.Fatalf("failed to create blob pool: %v", err)
}
defer pool.Close()
@ -817,8 +817,8 @@ func TestOpenIndex(t *testing.T) {
blobfee: uint256.NewInt(params.BlobTxMinBlobGasprice),
statedb: statedb,
}
pool := New(Config{Datadir: storage}, chain)
if err := pool.Init(1, chain.CurrentBlock(), makeAddressReserver()); err != nil {
pool := New(Config{Datadir: storage}, chain, nil)
if err := pool.Init(1, chain.CurrentBlock(), makeAddressReserver(), nil); err != nil {
t.Fatalf("failed to create blob pool: %v", err)
}
defer pool.Close()
@ -918,8 +918,8 @@ func TestOpenHeap(t *testing.T) {
blobfee: uint256.NewInt(105),
statedb: statedb,
}
pool := New(Config{Datadir: storage}, chain)
if err := pool.Init(1, chain.CurrentBlock(), makeAddressReserver()); err != nil {
pool := New(Config{Datadir: storage}, chain, nil)
if err := pool.Init(1, chain.CurrentBlock(), makeAddressReserver(), nil); err != nil {
t.Fatalf("failed to create blob pool: %v", err)
}
defer pool.Close()
@ -997,8 +997,8 @@ func TestOpenCap(t *testing.T) {
blobfee: uint256.NewInt(105),
statedb: statedb,
}
pool := New(Config{Datadir: storage, Datacap: datacap}, chain)
if err := pool.Init(1, chain.CurrentBlock(), makeAddressReserver()); err != nil {
pool := New(Config{Datadir: storage, Datacap: datacap}, chain, nil)
if err := pool.Init(1, chain.CurrentBlock(), makeAddressReserver(), nil); err != nil {
t.Fatalf("failed to create blob pool: %v", err)
}
// Verify that enough transactions have been dropped to get the pool's size
@ -1098,8 +1098,8 @@ func TestChangingSlotterSize(t *testing.T) {
blobfee: uint256.NewInt(105),
statedb: statedb,
}
pool := New(Config{Datadir: storage}, chain)
if err := pool.Init(1, chain.CurrentBlock(), makeAddressReserver()); err != nil {
pool := New(Config{Datadir: storage}, chain, nil)
if err := pool.Init(1, chain.CurrentBlock(), makeAddressReserver(), nil); err != nil {
t.Fatalf("failed to create blob pool: %v", err)
}
@ -1541,8 +1541,8 @@ func TestAdd(t *testing.T) {
blobfee: uint256.NewInt(105),
statedb: statedb,
}
pool := New(Config{Datadir: storage}, chain)
if err := pool.Init(1, chain.CurrentBlock(), makeAddressReserver()); err != nil {
pool := New(Config{Datadir: storage}, chain, nil)
if err := pool.Init(1, chain.CurrentBlock(), makeAddressReserver(), nil); err != nil {
t.Fatalf("test %d: failed to create blob pool: %v", i, err)
}
verifyPoolInternals(t, pool)
@ -1638,10 +1638,10 @@ func benchmarkPoolPending(b *testing.B, datacap uint64) {
blobfee: uint256.NewInt(blobfee),
statedb: statedb,
}
pool = New(Config{Datadir: ""}, chain)
pool = New(Config{Datadir: ""}, chain, nil)
)
if err := pool.Init(1, chain.CurrentBlock(), makeAddressReserver()); err != nil {
if err := pool.Init(1, chain.CurrentBlock(), makeAddressReserver(), nil); err != nil {
b.Fatalf("failed to create blob pool: %v", err)
}
// Make the pool not use disk (just drop everything). This test never reads

4
core/txpool/errors.go
View file

@ -56,4 +56,8 @@ var (
// input transaction of non-blob type when a blob transaction from this sender
// 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")
)

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

@ -63,10 +63,6 @@ var (
// another remote transaction.
ErrTxPoolOverflow = errors.New("txpool is full")
// 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")
// 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")
@ -243,6 +239,8 @@ type LegacyPool struct {
pendingNonces *noncer // Pending state tracking virtual nonces
reserve txpool.AddressReserver // Address reserver to ensure exclusivity across subpools
isReserved txpool.IsAddressReserved // Address reserver to check whether the address has been reserved
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
@ -306,9 +304,10 @@ func (pool *LegacyPool) Filter(tx *types.Transaction) bool {
// Init sets the gas price needed to keep a transaction in the pool and the chain
// head to allow balance / nonce checks. The internal
// goroutines will be spun up and the pool deemed operational afterwards.
func (pool *LegacyPool) Init(gasTip uint64, head *types.Header, reserve txpool.AddressReserver) error {
func (pool *LegacyPool) Init(gasTip uint64, head *types.Header, reserve txpool.AddressReserver, isReserved txpool.IsAddressReserved) error {
// Set the address reserver to request exclusive access to pooled accounts
pool.reserve = reserve
pool.isReserved = isReserved
// Set the basic pool parameters
pool.gasTip.Store(uint256.NewInt(gasTip))
@ -618,7 +617,7 @@ func (pool *LegacyPool) checkDelegationLimit(tx *types.Transaction) error {
from, _ := types.Sender(pool.signer, tx) // validated
// Short circuit if the sender has neither delegation nor pending delegation.
if pool.currentState.GetCodeHash(from) == types.EmptyCodeHash && pool.all.delegationTxsCount(from) == 0 {
if pool.currentState.GetCodeHash(from) == types.EmptyCodeHash && !pool.all.hasAuth(from) {
return nil
}
pending := pool.pending[from]
@ -633,7 +632,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
@ -644,12 +643,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.isReserved(auth) {
return ErrAuthorityReserved
}
}
}
return nil
@ -1862,11 +1873,13 @@ func (t *lookup) removeAuthorities(tx *types.Transaction) {
}
}
// delegationTxsCount returns the number of pending authorizations for the specified address.
func (t *lookup) delegationTxsCount(addr common.Address) int {
// hasAuth returns a flag indicating whether there are pending authorizations
// from the specified address.
func (t *lookup) hasAuth(addr common.Address) bool {
t.lock.RLock()
defer t.lock.RUnlock()
return len(t.auths[addr])
return len(t.auths[addr]) > 0
}
// numSlots calculates the number of slots needed for a single transaction.
@ -1904,3 +1917,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)
}

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

@ -86,7 +86,9 @@ func TestTransactionFutureAttack(t *testing.T) {
config.GlobalQueue = 100
config.GlobalSlots = 100
pool := New(config, blockchain)
pool.Init(config.PriceLimit, blockchain.CurrentBlock(), makeAddressReserver())
reserver, isReserved := makeAddressReserver()
pool.Init(config.PriceLimit, blockchain.CurrentBlock(), reserver, isReserved)
defer pool.Close()
fillPool(t, pool)
pending, _ := pool.Stats()
@ -120,7 +122,9 @@ func TestTransactionFuture1559(t *testing.T) {
statedb, _ := state.New(types.EmptyRootHash, state.NewDatabaseForTesting())
blockchain := newTestBlockChain(eip1559Config, 1000000, statedb, new(event.Feed))
pool := New(testTxPoolConfig, blockchain)
pool.Init(testTxPoolConfig.PriceLimit, blockchain.CurrentBlock(), makeAddressReserver())
reserver, isReserved := makeAddressReserver()
pool.Init(testTxPoolConfig.PriceLimit, blockchain.CurrentBlock(), reserver, isReserved)
defer pool.Close()
// Create a number of test accounts, fund them and make transactions
@ -153,7 +157,9 @@ func TestTransactionZAttack(t *testing.T) {
statedb, _ := state.New(types.EmptyRootHash, state.NewDatabaseForTesting())
blockchain := newTestBlockChain(eip1559Config, 1000000, statedb, new(event.Feed))
pool := New(testTxPoolConfig, blockchain)
pool.Init(testTxPoolConfig.PriceLimit, blockchain.CurrentBlock(), makeAddressReserver())
reserver, isReserved := makeAddressReserver()
pool.Init(testTxPoolConfig.PriceLimit, blockchain.CurrentBlock(), reserver, isReserved)
defer pool.Close()
// Create a number of test accounts, fund them and make transactions
fillPool(t, pool)
@ -224,7 +230,9 @@ func BenchmarkFutureAttack(b *testing.B) {
config.GlobalQueue = 100
config.GlobalSlots = 100
pool := New(config, blockchain)
pool.Init(testTxPoolConfig.PriceLimit, blockchain.CurrentBlock(), makeAddressReserver())
reserver, isReserved := makeAddressReserver()
pool.Init(testTxPoolConfig.PriceLimit, blockchain.CurrentBlock(), reserver, isReserved)
defer pool.Close()
fillPool(b, pool)

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

@ -168,7 +168,7 @@ func pricedSetCodeTxWithAuth(nonce uint64, gaslimit uint64, gasFee, tip *uint256
})
}
func makeAddressReserver() txpool.AddressReserver {
func makeAddressReserver() (txpool.AddressReserver, txpool.IsAddressReserved) {
var (
reserved = make(map[common.Address]struct{})
lock sync.Mutex
@ -190,6 +190,12 @@ func makeAddressReserver() txpool.AddressReserver {
}
delete(reserved, addr)
return nil
}, func(addr common.Address) bool {
lock.Lock()
defer lock.Unlock()
_, exists := reserved[addr]
return exists
}
}
@ -203,7 +209,9 @@ func setupPoolWithConfig(config *params.ChainConfig) (*LegacyPool, *ecdsa.Privat
key, _ := crypto.GenerateKey()
pool := New(testTxPoolConfig, blockchain)
if err := pool.Init(testTxPoolConfig.PriceLimit, blockchain.CurrentBlock(), makeAddressReserver()); err != nil {
reserver, isReserved := makeAddressReserver()
if err := pool.Init(testTxPoolConfig.PriceLimit, blockchain.CurrentBlock(), reserver, isReserved); err != nil {
panic(err)
}
// wait for the pool to initialize
@ -336,7 +344,9 @@ func TestStateChangeDuringReset(t *testing.T) {
tx1 := transaction(1, 100000, key)
pool := New(testTxPoolConfig, blockchain)
pool.Init(testTxPoolConfig.PriceLimit, blockchain.CurrentBlock(), makeAddressReserver())
reserver, isReserved := makeAddressReserver()
pool.Init(testTxPoolConfig.PriceLimit, blockchain.CurrentBlock(), reserver, isReserved)
defer pool.Close()
nonce := pool.Nonce(address)
@ -753,7 +763,9 @@ func TestPostponing(t *testing.T) {
blockchain := newTestBlockChain(params.TestChainConfig, 1000000, statedb, new(event.Feed))
pool := New(testTxPoolConfig, blockchain)
pool.Init(testTxPoolConfig.PriceLimit, blockchain.CurrentBlock(), makeAddressReserver())
reserver, isReserved := makeAddressReserver()
pool.Init(testTxPoolConfig.PriceLimit, blockchain.CurrentBlock(), reserver, isReserved)
defer pool.Close()
// Create two test accounts to produce different gap profiles with
@ -963,7 +975,9 @@ func TestQueueGlobalLimiting(t *testing.T) {
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(testTxPoolConfig.PriceLimit, blockchain.CurrentBlock(), makeAddressReserver())
reserver, isReserved := makeAddressReserver()
pool.Init(testTxPoolConfig.PriceLimit, blockchain.CurrentBlock(), reserver, isReserved)
defer pool.Close()
// Create a number of test accounts and fund them (last one will be the local)
@ -1015,7 +1029,9 @@ func TestQueueTimeLimiting(t *testing.T) {
config.Lifetime = time.Second
pool := New(config, blockchain)
pool.Init(config.PriceLimit, blockchain.CurrentBlock(), makeAddressReserver())
reserver, isReserved := makeAddressReserver()
pool.Init(config.PriceLimit, blockchain.CurrentBlock(), reserver, isReserved)
defer pool.Close()
// Create a test account to ensure remotes expire
@ -1176,7 +1192,9 @@ func TestPendingGlobalLimiting(t *testing.T) {
config.GlobalSlots = config.AccountSlots * 10
pool := New(config, blockchain)
pool.Init(config.PriceLimit, blockchain.CurrentBlock(), makeAddressReserver())
reserver, isReserved := makeAddressReserver()
pool.Init(config.PriceLimit, blockchain.CurrentBlock(), reserver, isReserved)
defer pool.Close()
// Create a number of test accounts and fund them
@ -1275,7 +1293,9 @@ func TestCapClearsFromAll(t *testing.T) {
config.GlobalSlots = 8
pool := New(config, blockchain)
pool.Init(config.PriceLimit, blockchain.CurrentBlock(), makeAddressReserver())
reserver, isReserved := makeAddressReserver()
pool.Init(config.PriceLimit, blockchain.CurrentBlock(), reserver, isReserved)
defer pool.Close()
// Create a number of test accounts and fund them
@ -1308,7 +1328,9 @@ func TestPendingMinimumAllowance(t *testing.T) {
config.GlobalSlots = 1
pool := New(config, blockchain)
pool.Init(config.PriceLimit, blockchain.CurrentBlock(), makeAddressReserver())
reserver, isReserved := makeAddressReserver()
pool.Init(config.PriceLimit, blockchain.CurrentBlock(), reserver, isReserved)
defer pool.Close()
// Create a number of test accounts and fund them
@ -1352,7 +1374,9 @@ func TestRepricing(t *testing.T) {
blockchain := newTestBlockChain(params.TestChainConfig, 1000000, statedb, new(event.Feed))
pool := New(testTxPoolConfig, blockchain)
pool.Init(testTxPoolConfig.PriceLimit, blockchain.CurrentBlock(), makeAddressReserver())
reserver, isReserved := makeAddressReserver()
pool.Init(testTxPoolConfig.PriceLimit, blockchain.CurrentBlock(), reserver, isReserved)
defer pool.Close()
// Keep track of transaction events to ensure all executables get announced
@ -1457,7 +1481,9 @@ func TestMinGasPriceEnforced(t *testing.T) {
txPoolConfig := DefaultConfig
txPoolConfig.NoLocals = true
pool := New(txPoolConfig, blockchain)
pool.Init(txPoolConfig.PriceLimit, blockchain.CurrentBlock(), makeAddressReserver())
reserver, isReserved := makeAddressReserver()
pool.Init(txPoolConfig.PriceLimit, blockchain.CurrentBlock(), reserver, isReserved)
defer pool.Close()
key, _ := crypto.GenerateKey()
@ -1606,7 +1632,9 @@ func TestUnderpricing(t *testing.T) {
config.GlobalQueue = 2
pool := New(config, blockchain)
pool.Init(config.PriceLimit, blockchain.CurrentBlock(), makeAddressReserver())
reserver, isReserved := makeAddressReserver()
pool.Init(config.PriceLimit, blockchain.CurrentBlock(), reserver, isReserved)
defer pool.Close()
// Keep track of transaction events to ensure all executables get announced
@ -1696,7 +1724,9 @@ func TestStableUnderpricing(t *testing.T) {
config.GlobalQueue = 0
pool := New(config, blockchain)
pool.Init(config.PriceLimit, blockchain.CurrentBlock(), makeAddressReserver())
reserver, isReserved := makeAddressReserver()
pool.Init(config.PriceLimit, blockchain.CurrentBlock(), reserver, isReserved)
defer pool.Close()
// Keep track of transaction events to ensure all executables get announced
@ -1899,7 +1929,9 @@ func TestDeduplication(t *testing.T) {
blockchain := newTestBlockChain(params.TestChainConfig, 1000000, statedb, new(event.Feed))
pool := New(testTxPoolConfig, blockchain)
pool.Init(testTxPoolConfig.PriceLimit, blockchain.CurrentBlock(), makeAddressReserver())
reserver, isReserved := makeAddressReserver()
pool.Init(testTxPoolConfig.PriceLimit, blockchain.CurrentBlock(), reserver, isReserved)
defer pool.Close()
// Create a test account to add transactions with
@ -1966,7 +1998,9 @@ func TestReplacement(t *testing.T) {
blockchain := newTestBlockChain(params.TestChainConfig, 1000000, statedb, new(event.Feed))
pool := New(testTxPoolConfig, blockchain)
pool.Init(testTxPoolConfig.PriceLimit, blockchain.CurrentBlock(), makeAddressReserver())
reserver, isReserved := makeAddressReserver()
pool.Init(testTxPoolConfig.PriceLimit, blockchain.CurrentBlock(), reserver, isReserved)
defer pool.Close()
// Keep track of transaction events to ensure all executables get announced
@ -2157,7 +2191,9 @@ func TestStatusCheck(t *testing.T) {
blockchain := newTestBlockChain(params.TestChainConfig, 1000000, statedb, new(event.Feed))
pool := New(testTxPoolConfig, blockchain)
pool.Init(testTxPoolConfig.PriceLimit, blockchain.CurrentBlock(), makeAddressReserver())
reserver, isReserved := makeAddressReserver()
pool.Init(testTxPoolConfig.PriceLimit, blockchain.CurrentBlock(), reserver, isReserved)
defer pool.Close()
// Create the test accounts to check various transaction statuses with
@ -2230,7 +2266,9 @@ func TestSetCodeTransactions(t *testing.T) {
blockchain := newTestBlockChain(params.MergedTestChainConfig, 1000000, statedb, new(event.Feed))
pool := New(testTxPoolConfig, blockchain)
pool.Init(testTxPoolConfig.PriceLimit, blockchain.CurrentBlock(), makeAddressReserver())
reserver, isReserved := makeAddressReserver()
pool.Init(testTxPoolConfig.PriceLimit, blockchain.CurrentBlock(), reserver, isReserved)
defer pool.Close()
// Create the test accounts
@ -2271,12 +2309,12 @@ func TestSetCodeTransactions(t *testing.T) {
if err := pool.addRemoteSync(pricedTransaction(0, 100000, big.NewInt(1), keyA)); err != nil {
t.Fatalf("%s: failed to add remote transaction: %v", name, err)
}
if err := pool.addRemoteSync(pricedTransaction(1, 100000, big.NewInt(1), keyA)); !errors.Is(err, ErrInflightTxLimitReached) {
t.Fatalf("%s: error mismatch: want %v, have %v", name, ErrInflightTxLimitReached, err)
if err := pool.addRemoteSync(pricedTransaction(1, 100000, big.NewInt(1), 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, big.NewInt(1), keyA)); !errors.Is(err, ErrInflightTxLimitReached) {
t.Fatalf("%s: error mismatch: want %v, have %v", name, ErrInflightTxLimitReached, err)
if err := pool.addRemoteSync(pricedTransaction(2, 100000, big.NewInt(1), 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, big.NewInt(10), keyA)); err != nil {
@ -2310,8 +2348,8 @@ func TestSetCodeTransactions(t *testing.T) {
t.Fatalf("%s: failed to add with pending delegatio: %v", name, err)
}
// Also check gapped transaction is rejected.
if err := pool.addRemoteSync(pricedTransaction(1, 100000, big.NewInt(1), keyC)); !errors.Is(err, ErrInflightTxLimitReached) {
t.Fatalf("%s: error mismatch: want %v, have %v", name, ErrInflightTxLimitReached, err)
if err := pool.addRemoteSync(pricedTransaction(1, 100000, big.NewInt(1), keyC)); !errors.Is(err, txpool.ErrInflightTxLimitReached) {
t.Fatalf("%s: error mismatch: want %v, have %v", name, txpool.ErrInflightTxLimitReached, err)
}
},
},
@ -2386,7 +2424,7 @@ func TestSetCodeTransactions(t *testing.T) {
if err := pool.addRemoteSync(pricedTransaction(0, 100000, big.NewInt(1000), 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, big.NewInt(1000), keyC)), ErrInflightTxLimitReached; !errors.Is(err, want) {
if err, want := pool.addRemoteSync(pricedTransaction(1, 100000, big.NewInt(1000), keyC)), txpool.ErrInflightTxLimitReached; !errors.Is(err, want) {
t.Fatalf("%s: error mismatch: want %v, have %v", name, want, err)
}
},
@ -2454,7 +2492,9 @@ func TestSetCodeTransactionsReorg(t *testing.T) {
blockchain := newTestBlockChain(params.MergedTestChainConfig, 1000000, statedb, new(event.Feed))
pool := New(testTxPoolConfig, blockchain)
pool.Init(testTxPoolConfig.PriceLimit, blockchain.CurrentBlock(), makeAddressReserver())
reserver, isReserved := makeAddressReserver()
pool.Init(testTxPoolConfig.PriceLimit, blockchain.CurrentBlock(), reserver, isReserved)
defer pool.Close()
// Create the test accounts
@ -2489,8 +2529,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, big.NewInt(1000), keyA)); !errors.Is(err, ErrInflightTxLimitReached) {
t.Fatalf("unexpected error %v, expecting %v", err, ErrInflightTxLimitReached)
if err := pool.addRemoteSync(pricedTransaction(2, 100000, big.NewInt(1000), 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, tracing.NonceChangeAuthorization)

6
core/txpool/subpool.go
View file

@ -71,6 +71,10 @@ type LazyResolver interface {
// may request (and relinquish) exclusive access to certain addresses.
type AddressReserver func(addr common.Address, reserve bool) error
// IsAddressReserved is passed by the main transaction pool to subpools, so they
// can query whether the address has been reserved or not.
type IsAddressReserved func(addr common.Address) bool
// PendingFilter is a collection of filter rules to allow retrieving a subset
// of transactions for announcement or mining.
//
@ -109,7 +113,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 uint64, head *types.Header, reserve AddressReserver) error
Init(gasTip uint64, head *types.Header, reserve AddressReserver, isReserved IsAddressReserved) error
// Close terminates any background processing threads and releases any held
// resources.

13
core/txpool/txpool.go
View file

@ -82,7 +82,7 @@ type TxPool struct {
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
reserveLock sync.RWMutex // 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
@ -120,7 +120,7 @@ func New(gasTip uint64, chain BlockChain, subpools []SubPool) (*TxPool, error) {
sync: make(chan chan error),
}
for i, subpool := range subpools {
if err := subpool.Init(gasTip, head, pool.reserver(i, subpool)); err != nil {
if err := subpool.Init(gasTip, head, pool.reserver(i, subpool), pool.isReserved); err != nil {
for j := i - 1; j >= 0; j-- {
subpools[j].Close()
}
@ -131,6 +131,15 @@ func New(gasTip uint64, chain BlockChain, subpools []SubPool) (*TxPool, error) {
return pool, nil
}
// isReserved returns a flag indicating if the address has been reserved or not.
func (p *TxPool) isReserved(address common.Address) bool {
p.reserveLock.RLock()
defer p.reserveLock.RUnlock()
_, exists := p.reservations[address]
return exists
}
// 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

10
eth/backend.go
View file

@ -260,16 +260,16 @@ func New(stack *node.Node, config *ethconfig.Config) (*Ethereum, error) {
eth.filterMaps = filtermaps.NewFilterMaps(chainDb, chainView, historyCutoff, finalBlock, filtermaps.DefaultParams, fmConfig)
eth.closeFilterMaps = make(chan chan struct{})
if config.BlobPool.Datadir != "" {
config.BlobPool.Datadir = stack.ResolvePath(config.BlobPool.Datadir)
}
blobPool := blobpool.New(config.BlobPool, eth.blockchain)
if config.TxPool.Journal != "" {
config.TxPool.Journal = stack.ResolvePath(config.TxPool.Journal)
}
legacyPool := legacypool.New(config.TxPool, eth.blockchain)
if config.BlobPool.Datadir != "" {
config.BlobPool.Datadir = stack.ResolvePath(config.BlobPool.Datadir)
}
blobPool := blobpool.New(config.BlobPool, eth.blockchain, legacyPool.HasPendingAuth)
eth.txPool, err = txpool.New(config.TxPool.PriceLimit, eth.blockchain, []txpool.SubPool{legacyPool, blobPool})
if err != nil {
return nil, err

2
eth/protocols/eth/handler_test.go
View file

@ -135,7 +135,7 @@ func newTestBackendWithGenerator(blocks int, shanghai bool, cancun bool, generat
storage, _ := os.MkdirTemp("", "blobpool-")
defer os.RemoveAll(storage)
blobPool := blobpool.New(blobpool.Config{Datadir: storage}, chain)
blobPool := blobpool.New(blobpool.Config{Datadir: storage}, chain, nil)
legacyPool := legacypool.New(txconfig, chain)
txpool, _ := txpool.New(txconfig.PriceLimit, chain, []txpool.SubPool{legacyPool, blobPool})