mirror of
https://github.com/ethereum/go-ethereum.git
synced 2026-07-27 07:06:42 +00:00
core/txpool: introduce reserver
This commit is contained in:
Gary Rong
parent
651db00cdf
commit
710bcc3d80
8 changed files with 174 additions and 241 deletions
|
|
@ -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)
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
109
core/txpool/reserver.go
Normal file
109
core/txpool/reserver.go
Normal file
|
|
@ -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 <http://www.gnu.org/licenses/>.
|
||||
|
||||
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
|
||||
}
|
||||
|
|
@ -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.
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
Loading…
Reference in a new issue