diff --git a/core/txpool/blobpool/blobpool.go b/core/txpool/blobpool/blobpool.go index b299a5da4e..c14e560b90 100644 --- a/core/txpool/blobpool/blobpool.go +++ b/core/txpool/blobpool/blobpool.go @@ -47,6 +47,8 @@ import ( ) const ( + poolName = "blobpool" // The name of legacy txpool + // blobSize is the protocol constrained byte size of a single blob in a // transaction. There can be multiple of these embedded into a single tx. blobSize = params.BlobTxFieldElementsPerBlob * params.BlobTxBytesPerFieldElement @@ -299,7 +301,7 @@ func newBlobTxMeta(id uint64, size uint64, storageSize uint32, tx *types.Transac // and leading up to the first no-change. type BlobPool struct { config Config // Pool configuration - reserve txpool.AddressReserver // Address reserver to ensure exclusivity across subpools + reserver *txpool.Reserver // 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 @@ -355,8 +357,8 @@ 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, _ txpool.IsAddressReserved) error { - p.reserve = reserve +func (p *BlobPool) Init(gasTip uint64, head *types.Header, reserver *txpool.Reserver) error { + p.reserver = reserver var ( queuedir string @@ -501,7 +503,7 @@ func (p *BlobPool) parseTransaction(id uint64, size uint32, blob []byte) error { return err } if _, ok := p.index[sender]; !ok { - if err := p.reserve(sender, true); err != nil { + if err := p.reserver.Hold(sender, poolName); err != nil { return err } p.index[sender] = []*blobTxMeta{} @@ -556,7 +558,7 @@ func (p *BlobPool) recheck(addr common.Address, inclusions map[common.Hash]uint6 if inclusions != nil { // only during reorgs will the heap be initialized heap.Remove(p.evict, p.evict.index[addr]) } - p.reserve(addr, false) + p.reserver.Release(addr, poolName) if gapped { log.Warn("Dropping dangling blob transactions", "from", addr, "missing", next, "drop", nonces, "ids", ids) @@ -709,7 +711,7 @@ func (p *BlobPool) recheck(addr common.Address, inclusions map[common.Hash]uint6 if inclusions != nil { // only during reorgs will the heap be initialized heap.Remove(p.evict, p.evict.index[addr]) } - p.reserve(addr, false) + p.reserver.Release(addr, poolName) } else { p.index[addr] = txs } @@ -1008,7 +1010,7 @@ func (p *BlobPool) reinject(addr common.Address, txhash common.Hash) error { // Update the indices and metrics meta := newBlobTxMeta(id, tx.Size(), p.store.Size(id), tx) if _, ok := p.index[addr]; !ok { - if err := p.reserve(addr, true); err != nil { + if err := p.reserver.Hold(addr, poolName); err != nil { log.Warn("Failed to reserve account for blob pool", "tx", tx.Hash(), "from", addr, "err", err) return err } @@ -1068,7 +1070,7 @@ func (p *BlobPool) SetGasTip(tip *big.Int) { delete(p.spent, addr) heap.Remove(p.evict, p.evict.index[addr]) - p.reserve(addr, false) + p.reserver.Release(addr, poolName) } // Clear out the transactions from the data store log.Warn("Dropping underpriced blob transaction", "from", addr, "rejected", tx.nonce, "tip", tx.execTipCap, "want", tip, "drop", nonces, "ids", ids) @@ -1407,7 +1409,7 @@ func (p *BlobPool) add(tx *types.Transaction) (err error) { // only by this subpool until all transactions are evicted from, _ := types.Sender(p.signer, tx) // already validated above if _, ok := p.index[from]; !ok { - if err := p.reserve(from, true); err != nil { + if err := p.reserver.Hold(from, poolName); err != nil { addNonExclusiveMeter.Mark(1) return err } @@ -1419,7 +1421,7 @@ func (p *BlobPool) add(tx *types.Transaction) (err error) { // by a return statement before running deferred methods. Take care with // removing or subscoping err as it will break this clause. if err != nil { - p.reserve(from, false) + p.reserver.Release(from, poolName) } }() } @@ -1551,7 +1553,7 @@ func (p *BlobPool) drop() { if last { delete(p.index, from) delete(p.spent, from) - p.reserve(from, false) + p.reserver.Release(from, poolName) } else { txs[len(txs)-1] = nil txs = txs[:len(txs)-1] @@ -1827,7 +1829,7 @@ func (p *BlobPool) Clear() { // can't happen until Clear releases the reservation lock. Clear cannot // acquire the subpool lock until the transaction addition is completed. for acct := range p.index { - p.reserve(acct, false) + p.reserver.Release(acct, poolName) } p.lookup = newLookup() p.index = make(map[common.Address][]*blobTxMeta) diff --git a/core/txpool/blobpool/blobpool_test.go b/core/txpool/blobpool/blobpool_test.go index 884b9d468d..1989100761 100644 --- a/core/txpool/blobpool/blobpool_test.go +++ b/core/txpool/blobpool/blobpool_test.go @@ -26,7 +26,6 @@ import ( "math/big" "os" "path/filepath" - "sync" "testing" "github.com/ethereum/go-ethereum/common" @@ -168,33 +167,6 @@ func (bc *testBlockChain) StateAt(common.Hash) (*state.StateDB, error) { return bc.statedb, nil } -// makeAddressReserver is a utility method to sanity check that accounts are -// properly reserved by the blobpool (no duplicate reserves or unreserves). -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 - } -} - // makeTx is a utility method to construct a random blob transaction and sign it // with a valid key, only setting the interesting fields from the perspective of // the blob pool. @@ -700,7 +672,7 @@ func TestOpenDrops(t *testing.T) { statedb: statedb, } pool := New(Config{Datadir: storage}, chain, nil) - if err := pool.Init(1, chain.CurrentBlock(), makeAddressReserver(), nil); err != nil { + if err := pool.Init(1, chain.CurrentBlock(), txpool.NewReserver()); err != nil { t.Fatalf("failed to create blob pool: %v", err) } defer pool.Close() @@ -818,7 +790,7 @@ func TestOpenIndex(t *testing.T) { statedb: statedb, } pool := New(Config{Datadir: storage}, chain, nil) - if err := pool.Init(1, chain.CurrentBlock(), makeAddressReserver(), nil); err != nil { + if err := pool.Init(1, chain.CurrentBlock(), txpool.NewReserver()); err != nil { t.Fatalf("failed to create blob pool: %v", err) } defer pool.Close() @@ -919,7 +891,7 @@ func TestOpenHeap(t *testing.T) { statedb: statedb, } pool := New(Config{Datadir: storage}, chain, nil) - if err := pool.Init(1, chain.CurrentBlock(), makeAddressReserver(), nil); err != nil { + if err := pool.Init(1, chain.CurrentBlock(), txpool.NewReserver()); err != nil { t.Fatalf("failed to create blob pool: %v", err) } defer pool.Close() @@ -998,7 +970,7 @@ func TestOpenCap(t *testing.T) { statedb: statedb, } pool := New(Config{Datadir: storage, Datacap: datacap}, chain, nil) - if err := pool.Init(1, chain.CurrentBlock(), makeAddressReserver(), nil); err != nil { + if err := pool.Init(1, chain.CurrentBlock(), txpool.NewReserver()); err != nil { t.Fatalf("failed to create blob pool: %v", err) } // Verify that enough transactions have been dropped to get the pool's size @@ -1099,7 +1071,7 @@ func TestChangingSlotterSize(t *testing.T) { statedb: statedb, } pool := New(Config{Datadir: storage}, chain, nil) - if err := pool.Init(1, chain.CurrentBlock(), makeAddressReserver(), nil); err != nil { + if err := pool.Init(1, chain.CurrentBlock(), txpool.NewReserver()); err != nil { t.Fatalf("failed to create blob pool: %v", err) } @@ -1542,7 +1514,7 @@ func TestAdd(t *testing.T) { statedb: statedb, } pool := New(Config{Datadir: storage}, chain, nil) - if err := pool.Init(1, chain.CurrentBlock(), makeAddressReserver(), nil); err != nil { + if err := pool.Init(1, chain.CurrentBlock(), txpool.NewReserver()); err != nil { t.Fatalf("test %d: failed to create blob pool: %v", i, err) } verifyPoolInternals(t, pool) @@ -1641,7 +1613,7 @@ func benchmarkPoolPending(b *testing.B, datacap uint64) { pool = New(Config{Datadir: ""}, chain, nil) ) - if err := pool.Init(1, chain.CurrentBlock(), makeAddressReserver(), nil); err != nil { + if err := pool.Init(1, chain.CurrentBlock(), txpool.NewReserver()); err != nil { b.Fatalf("failed to create blob pool: %v", err) } // Make the pool not use disk (just drop everything). This test never reads diff --git a/core/txpool/legacypool/legacypool.go b/core/txpool/legacypool/legacypool.go index ff9159d967..597dc3cdca 100644 --- a/core/txpool/legacypool/legacypool.go +++ b/core/txpool/legacypool/legacypool.go @@ -45,6 +45,8 @@ import ( ) const ( + poolName = "general" // The name of legacy txpool + // txSlotSize is used to calculate how many data slots a single transaction // takes up based on its size. The slots are used as DoS protection, ensuring // that validating a new transaction remains a constant operation (in reality @@ -237,9 +239,7 @@ 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 - - reserve txpool.AddressReserver // Address reserver to ensure exclusivity across subpools - isReserved txpool.IsAddressReserved // Address reserver to check whether the address has been reserved + reserver *txpool.Reserver // 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 @@ -304,10 +304,9 @@ 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, isReserved txpool.IsAddressReserved) error { +func (pool *LegacyPool) Init(gasTip uint64, head *types.Header, reserver *txpool.Reserver) error { // Set the address reserver to request exclusive access to pooled accounts - pool.reserve = reserve - pool.isReserved = isReserved + pool.reserver = reserver // Set the basic pool parameters pool.gasTip.Store(uint256.NewInt(gasTip)) @@ -658,7 +657,7 @@ func (pool *LegacyPool) validateAuth(tx *types.Transaction) error { // 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) { + if pool.reserver.Has(auth) { return ErrAuthorityReserved } } @@ -694,7 +693,7 @@ func (pool *LegacyPool) add(tx *types.Transaction) (replaced bool, err error) { _, hasQueued = pool.queue[from] ) if !hasPending && !hasQueued { - if err := pool.reserve(from, true); err != nil { + if err := pool.reserver.Hold(from, poolName); err != nil { return false, err } defer func() { @@ -705,7 +704,7 @@ func (pool *LegacyPool) add(tx *types.Transaction) (replaced bool, err error) { // 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, poolName) } }() } @@ -1098,7 +1097,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, poolName) } }() } @@ -1478,7 +1477,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, poolName) } } } @@ -1664,7 +1663,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, poolName) } } } @@ -1909,7 +1908,7 @@ func (pool *LegacyPool) Clear() { // acquire the subpool lock until the transaction addition is completed. for _, tx := range pool.all.txs { senderAddr, _ := types.Sender(pool.signer, tx) - pool.reserve(senderAddr, false) + pool.reserver.Release(senderAddr, poolName) } pool.all = newLookup() pool.priced = newPricedList(pool.all) diff --git a/core/txpool/legacypool/legacypool2_test.go b/core/txpool/legacypool/legacypool2_test.go index 7fb99705d7..c68fc3316d 100644 --- a/core/txpool/legacypool/legacypool2_test.go +++ b/core/txpool/legacypool/legacypool2_test.go @@ -24,6 +24,7 @@ import ( "github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/core/state" "github.com/ethereum/go-ethereum/core/tracing" + "github.com/ethereum/go-ethereum/core/txpool" "github.com/ethereum/go-ethereum/core/types" "github.com/ethereum/go-ethereum/crypto" "github.com/ethereum/go-ethereum/event" @@ -86,9 +87,7 @@ func TestTransactionFutureAttack(t *testing.T) { config.GlobalQueue = 100 config.GlobalSlots = 100 pool := New(config, blockchain) - - reserver, isReserved := makeAddressReserver() - pool.Init(config.PriceLimit, blockchain.CurrentBlock(), reserver, isReserved) + pool.Init(config.PriceLimit, blockchain.CurrentBlock(), txpool.NewReserver()) defer pool.Close() fillPool(t, pool) pending, _ := pool.Stats() @@ -122,9 +121,7 @@ func TestTransactionFuture1559(t *testing.T) { statedb, _ := state.New(types.EmptyRootHash, state.NewDatabaseForTesting()) blockchain := newTestBlockChain(eip1559Config, 1000000, statedb, new(event.Feed)) pool := New(testTxPoolConfig, blockchain) - - reserver, isReserved := makeAddressReserver() - pool.Init(testTxPoolConfig.PriceLimit, blockchain.CurrentBlock(), reserver, isReserved) + pool.Init(testTxPoolConfig.PriceLimit, blockchain.CurrentBlock(), txpool.NewReserver()) defer pool.Close() // Create a number of test accounts, fund them and make transactions @@ -157,9 +154,7 @@ func TestTransactionZAttack(t *testing.T) { statedb, _ := state.New(types.EmptyRootHash, state.NewDatabaseForTesting()) blockchain := newTestBlockChain(eip1559Config, 1000000, statedb, new(event.Feed)) pool := New(testTxPoolConfig, blockchain) - - reserver, isReserved := makeAddressReserver() - pool.Init(testTxPoolConfig.PriceLimit, blockchain.CurrentBlock(), reserver, isReserved) + pool.Init(testTxPoolConfig.PriceLimit, blockchain.CurrentBlock(), txpool.NewReserver()) defer pool.Close() // Create a number of test accounts, fund them and make transactions fillPool(t, pool) @@ -230,9 +225,7 @@ func BenchmarkFutureAttack(b *testing.B) { config.GlobalQueue = 100 config.GlobalSlots = 100 pool := New(config, blockchain) - - reserver, isReserved := makeAddressReserver() - pool.Init(testTxPoolConfig.PriceLimit, blockchain.CurrentBlock(), reserver, isReserved) + pool.Init(testTxPoolConfig.PriceLimit, blockchain.CurrentBlock(), txpool.NewReserver()) defer pool.Close() fillPool(b, pool) diff --git a/core/txpool/legacypool/legacypool_test.go b/core/txpool/legacypool/legacypool_test.go index c5f0702a88..83a94c8609 100644 --- a/core/txpool/legacypool/legacypool_test.go +++ b/core/txpool/legacypool/legacypool_test.go @@ -24,7 +24,6 @@ import ( "math/big" "math/rand" "slices" - "sync" "sync/atomic" "testing" "time" @@ -168,37 +167,6 @@ func pricedSetCodeTxWithAuth(nonce uint64, gaslimit uint64, gasFee, tip *uint256 }) } -func makeAddressReserver() (txpool.AddressReserver, txpool.IsAddressReserved) { - 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(addr common.Address) bool { - lock.Lock() - defer lock.Unlock() - - _, exists := reserved[addr] - return exists - } -} - func setupPool() (*LegacyPool, *ecdsa.PrivateKey) { return setupPoolWithConfig(params.TestChainConfig) } @@ -209,9 +177,7 @@ func setupPoolWithConfig(config *params.ChainConfig) (*LegacyPool, *ecdsa.Privat key, _ := crypto.GenerateKey() pool := New(testTxPoolConfig, blockchain) - - reserver, isReserved := makeAddressReserver() - if err := pool.Init(testTxPoolConfig.PriceLimit, blockchain.CurrentBlock(), reserver, isReserved); err != nil { + if err := pool.Init(testTxPoolConfig.PriceLimit, blockchain.CurrentBlock(), txpool.NewReserver()); err != nil { panic(err) } // wait for the pool to initialize @@ -344,9 +310,7 @@ func TestStateChangeDuringReset(t *testing.T) { tx1 := transaction(1, 100000, key) pool := New(testTxPoolConfig, blockchain) - - reserver, isReserved := makeAddressReserver() - pool.Init(testTxPoolConfig.PriceLimit, blockchain.CurrentBlock(), reserver, isReserved) + pool.Init(testTxPoolConfig.PriceLimit, blockchain.CurrentBlock(), txpool.NewReserver()) defer pool.Close() nonce := pool.Nonce(address) @@ -763,9 +727,7 @@ func TestPostponing(t *testing.T) { blockchain := newTestBlockChain(params.TestChainConfig, 1000000, statedb, new(event.Feed)) pool := New(testTxPoolConfig, blockchain) - - reserver, isReserved := makeAddressReserver() - pool.Init(testTxPoolConfig.PriceLimit, blockchain.CurrentBlock(), reserver, isReserved) + pool.Init(testTxPoolConfig.PriceLimit, blockchain.CurrentBlock(), txpool.NewReserver()) defer pool.Close() // Create two test accounts to produce different gap profiles with @@ -975,9 +937,7 @@ 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) - - reserver, isReserved := makeAddressReserver() - pool.Init(testTxPoolConfig.PriceLimit, blockchain.CurrentBlock(), reserver, isReserved) + pool.Init(testTxPoolConfig.PriceLimit, blockchain.CurrentBlock(), txpool.NewReserver()) defer pool.Close() // Create a number of test accounts and fund them (last one will be the local) @@ -1029,9 +989,7 @@ func TestQueueTimeLimiting(t *testing.T) { config.Lifetime = time.Second pool := New(config, blockchain) - - reserver, isReserved := makeAddressReserver() - pool.Init(config.PriceLimit, blockchain.CurrentBlock(), reserver, isReserved) + pool.Init(config.PriceLimit, blockchain.CurrentBlock(), txpool.NewReserver()) defer pool.Close() // Create a test account to ensure remotes expire @@ -1192,9 +1150,7 @@ func TestPendingGlobalLimiting(t *testing.T) { config.GlobalSlots = config.AccountSlots * 10 pool := New(config, blockchain) - - reserver, isReserved := makeAddressReserver() - pool.Init(config.PriceLimit, blockchain.CurrentBlock(), reserver, isReserved) + pool.Init(config.PriceLimit, blockchain.CurrentBlock(), txpool.NewReserver()) defer pool.Close() // Create a number of test accounts and fund them @@ -1293,9 +1249,7 @@ func TestCapClearsFromAll(t *testing.T) { config.GlobalSlots = 8 pool := New(config, blockchain) - - reserver, isReserved := makeAddressReserver() - pool.Init(config.PriceLimit, blockchain.CurrentBlock(), reserver, isReserved) + pool.Init(config.PriceLimit, blockchain.CurrentBlock(), txpool.NewReserver()) defer pool.Close() // Create a number of test accounts and fund them @@ -1328,9 +1282,7 @@ func TestPendingMinimumAllowance(t *testing.T) { config.GlobalSlots = 1 pool := New(config, blockchain) - - reserver, isReserved := makeAddressReserver() - pool.Init(config.PriceLimit, blockchain.CurrentBlock(), reserver, isReserved) + pool.Init(config.PriceLimit, blockchain.CurrentBlock(), txpool.NewReserver()) defer pool.Close() // Create a number of test accounts and fund them @@ -1374,9 +1326,7 @@ func TestRepricing(t *testing.T) { blockchain := newTestBlockChain(params.TestChainConfig, 1000000, statedb, new(event.Feed)) pool := New(testTxPoolConfig, blockchain) - - reserver, isReserved := makeAddressReserver() - pool.Init(testTxPoolConfig.PriceLimit, blockchain.CurrentBlock(), reserver, isReserved) + pool.Init(testTxPoolConfig.PriceLimit, blockchain.CurrentBlock(), txpool.NewReserver()) defer pool.Close() // Keep track of transaction events to ensure all executables get announced @@ -1481,9 +1431,7 @@ func TestMinGasPriceEnforced(t *testing.T) { txPoolConfig := DefaultConfig txPoolConfig.NoLocals = true pool := New(txPoolConfig, blockchain) - - reserver, isReserved := makeAddressReserver() - pool.Init(txPoolConfig.PriceLimit, blockchain.CurrentBlock(), reserver, isReserved) + pool.Init(txPoolConfig.PriceLimit, blockchain.CurrentBlock(), txpool.NewReserver()) defer pool.Close() key, _ := crypto.GenerateKey() @@ -1632,9 +1580,7 @@ func TestUnderpricing(t *testing.T) { config.GlobalQueue = 2 pool := New(config, blockchain) - - reserver, isReserved := makeAddressReserver() - pool.Init(config.PriceLimit, blockchain.CurrentBlock(), reserver, isReserved) + pool.Init(config.PriceLimit, blockchain.CurrentBlock(), txpool.NewReserver()) defer pool.Close() // Keep track of transaction events to ensure all executables get announced @@ -1724,9 +1670,7 @@ func TestStableUnderpricing(t *testing.T) { config.GlobalQueue = 0 pool := New(config, blockchain) - - reserver, isReserved := makeAddressReserver() - pool.Init(config.PriceLimit, blockchain.CurrentBlock(), reserver, isReserved) + pool.Init(config.PriceLimit, blockchain.CurrentBlock(), txpool.NewReserver()) defer pool.Close() // Keep track of transaction events to ensure all executables get announced @@ -1929,9 +1873,7 @@ func TestDeduplication(t *testing.T) { blockchain := newTestBlockChain(params.TestChainConfig, 1000000, statedb, new(event.Feed)) pool := New(testTxPoolConfig, blockchain) - - reserver, isReserved := makeAddressReserver() - pool.Init(testTxPoolConfig.PriceLimit, blockchain.CurrentBlock(), reserver, isReserved) + pool.Init(testTxPoolConfig.PriceLimit, blockchain.CurrentBlock(), txpool.NewReserver()) defer pool.Close() // Create a test account to add transactions with @@ -1998,9 +1940,7 @@ func TestReplacement(t *testing.T) { blockchain := newTestBlockChain(params.TestChainConfig, 1000000, statedb, new(event.Feed)) pool := New(testTxPoolConfig, blockchain) - - reserver, isReserved := makeAddressReserver() - pool.Init(testTxPoolConfig.PriceLimit, blockchain.CurrentBlock(), reserver, isReserved) + pool.Init(testTxPoolConfig.PriceLimit, blockchain.CurrentBlock(), txpool.NewReserver()) defer pool.Close() // Keep track of transaction events to ensure all executables get announced @@ -2191,9 +2131,7 @@ func TestStatusCheck(t *testing.T) { blockchain := newTestBlockChain(params.TestChainConfig, 1000000, statedb, new(event.Feed)) pool := New(testTxPoolConfig, blockchain) - - reserver, isReserved := makeAddressReserver() - pool.Init(testTxPoolConfig.PriceLimit, blockchain.CurrentBlock(), reserver, isReserved) + pool.Init(testTxPoolConfig.PriceLimit, blockchain.CurrentBlock(), txpool.NewReserver()) defer pool.Close() // Create the test accounts to check various transaction statuses with @@ -2266,9 +2204,7 @@ func TestSetCodeTransactions(t *testing.T) { blockchain := newTestBlockChain(params.MergedTestChainConfig, 1000000, statedb, new(event.Feed)) pool := New(testTxPoolConfig, blockchain) - - reserver, isReserved := makeAddressReserver() - pool.Init(testTxPoolConfig.PriceLimit, blockchain.CurrentBlock(), reserver, isReserved) + pool.Init(testTxPoolConfig.PriceLimit, blockchain.CurrentBlock(), txpool.NewReserver()) defer pool.Close() // Create the test accounts @@ -2492,9 +2428,7 @@ func TestSetCodeTransactionsReorg(t *testing.T) { blockchain := newTestBlockChain(params.MergedTestChainConfig, 1000000, statedb, new(event.Feed)) pool := New(testTxPoolConfig, blockchain) - - reserver, isReserved := makeAddressReserver() - pool.Init(testTxPoolConfig.PriceLimit, blockchain.CurrentBlock(), reserver, isReserved) + pool.Init(testTxPoolConfig.PriceLimit, blockchain.CurrentBlock(), txpool.NewReserver()) defer pool.Close() // Create the test accounts diff --git a/core/txpool/reserver.go b/core/txpool/reserver.go new file mode 100644 index 0000000000..6f107d7362 --- /dev/null +++ b/core/txpool/reserver.go @@ -0,0 +1,109 @@ +// 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 . + +package txpool + +import ( + "errors" + "fmt" + "sync" + + "github.com/ethereum/go-ethereum/common" + "github.com/ethereum/go-ethereum/log" + "github.com/ethereum/go-ethereum/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" +) + +// Reserver 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 Reserver struct { + accounts map[common.Address]string + lock sync.RWMutex +} + +// NewReserver initializes the account reserver. +func NewReserver() *Reserver { + return &Reserver{ + accounts: make(map[common.Address]string), + } +} + +// Hold attempts to reserve the specified account address for the given pool. +// Returns an error if the account is already reserved. +func (r *Reserver) Hold(addr common.Address, id string) error { + r.lock.Lock() + defer r.lock.Unlock() + + // Double reservations are forbidden even from the same pool to + // avoid subtle bugs in the long term. + owner, exists := r.accounts[addr] + if exists { + if owner == 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 + } + r.accounts[addr] = id + if metrics.Enabled() { + m := fmt.Sprintf("%s/%s", reservationsGaugeName, 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 (r *Reserver) Release(addr common.Address, id string) error { + r.lock.Lock() + defer r.lock.Unlock() + + // Ensure subpools only attempt to unreserve their own owned addresses, + // otherwise flag as a programming error. + owner, exists := r.accounts[addr] + if !exists { + log.Error("pool attempted to unreserve non-reserved address", "address", addr) + return errors.New("address not reserved") + } + if id != owner { + log.Error("pool attempted to unreserve non-owned address", "address", addr) + return errors.New("address not owned") + } + delete(r.accounts, addr) + if metrics.Enabled() { + m := fmt.Sprintf("%s/%s", reservationsGaugeName, id) + metrics.GetOrRegisterGauge(m, nil).Dec(1) + } + return nil +} + +// Has returns a flag indicating if the address has been reserved or not. +func (r *Reserver) Has(address common.Address) bool { + r.lock.RLock() + defer r.lock.RUnlock() + + _, exists := r.accounts[address] + return exists +} diff --git a/core/txpool/subpool.go b/core/txpool/subpool.go index 93e4763ae9..2cb5103875 100644 --- a/core/txpool/subpool.go +++ b/core/txpool/subpool.go @@ -67,14 +67,6 @@ type LazyResolver interface { Get(hash common.Hash) *types.Transaction } -// 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 - -// 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. // @@ -113,7 +105,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, isReserved IsAddressReserved) error + Init(gasTip uint64, head *types.Header, reserver *Reserver) error // Close terminates any background processing threads and releases any held // resources. diff --git a/core/txpool/txpool.go b/core/txpool/txpool.go index 1a8b684149..8016550ba3 100644 --- a/core/txpool/txpool.go +++ b/core/txpool/txpool.go @@ -29,7 +29,6 @@ import ( "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" ) @@ -43,15 +42,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 { @@ -81,9 +71,6 @@ type TxPool struct { 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.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 term chan struct{} // Termination channel to detect a closed pool @@ -110,17 +97,17 @@ func New(gasTip uint64, chain BlockChain, subpools []SubPool) (*TxPool, error) { 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{}), - sync: make(chan chan error), + subpools: subpools, + chain: chain, + signer: types.LatestSigner(chain.Config()), + state: statedb, + quit: make(chan chan error), + term: make(chan struct{}), + sync: make(chan chan error), } + reserver := NewReserver() for i, subpool := range subpools { - if err := subpool.Init(gasTip, head, pool.reserver(i, subpool), pool.isReserved); err != nil { + if err := subpool.Init(gasTip, head, reserver); err != nil { for j := i - 1; j >= 0; j-- { subpools[j].Close() } @@ -131,61 +118,6 @@ 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 -// 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