diff --git a/core/txpool/blobpool/blobpool.go b/core/txpool/blobpool/blobpool.go index 5a20c3ce5a..6159de6765 100644 --- a/core/txpool/blobpool/blobpool.go +++ b/core/txpool/blobpool/blobpool.go @@ -298,8 +298,9 @@ func newBlobTxMeta(id uint64, size uint64, storageSize uint32, tx *types.Transac // minimums will need to be done only starting at the swapped in/out nonce // 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 + 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 ( diff --git a/core/txpool/blobpool/blobpool_test.go b/core/txpool/blobpool/blobpool_test.go index b7c6cfa51e..884b9d468d 100644 --- a/core/txpool/blobpool/blobpool_test.go +++ b/core/txpool/blobpool/blobpool_test.go @@ -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 diff --git a/core/txpool/errors.go b/core/txpool/errors.go index c38644857e..02f5703b6c 100644 --- a/core/txpool/errors.go +++ b/core/txpool/errors.go @@ -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") ) diff --git a/core/txpool/legacypool/legacypool.go b/core/txpool/legacypool/legacypool.go index 9066f3e16b..ff9159d967 100644 --- a/core/txpool/legacypool/legacypool.go +++ b/core/txpool/legacypool/legacypool.go @@ -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") @@ -242,7 +238,9 @@ type LegacyPool struct { 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 + 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. @@ -1880,8 +1893,8 @@ func (pool *LegacyPool) Clear() { pool.mu.Lock() defer pool.mu.Unlock() - // unreserve each tracked account. Ideally, we could just clear the - // reservation map in the parent txpool context. However, if we clear in + // unreserve each tracked account. Ideally, we could just clear the + // reservation map in the parent txpool context. However, if we clear in // parent context, to avoid exposing the subpool lock, we have to lock the // reservations and then lock each subpool. // @@ -1892,7 +1905,7 @@ func (pool *LegacyPool) Clear() { // * TxPool.Clear attempts to lock subpool mutex // // The transaction addition may attempt to reserve the sender addr which - // can't happen until Clear releases the reservation lock. Clear cannot + // can't happen until Clear releases the reservation lock. Clear cannot // acquire the subpool lock until the transaction addition is completed. for _, tx := range pool.all.txs { senderAddr, _ := types.Sender(pool.signer, tx) @@ -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) +} diff --git a/core/txpool/legacypool/legacypool2_test.go b/core/txpool/legacypool/legacypool2_test.go index d55e85d74f..7fb99705d7 100644 --- a/core/txpool/legacypool/legacypool2_test.go +++ b/core/txpool/legacypool/legacypool2_test.go @@ -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) diff --git a/core/txpool/legacypool/legacypool_test.go b/core/txpool/legacypool/legacypool_test.go index 3f269bd69e..c5f0702a88 100644 --- a/core/txpool/legacypool/legacypool_test.go +++ b/core/txpool/legacypool/legacypool_test.go @@ -168,29 +168,35 @@ 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 ) return func(addr common.Address, reserve bool) error { - lock.Lock() - defer lock.Unlock() + lock.Lock() + defer lock.Unlock() - _, exists := reserved[addr] - if reserve { - if exists { - panic("already reserved") + _, exists := reserved[addr] + if reserve { + if exists { + panic("already reserved") + } + reserved[addr] = struct{}{} + return nil } - reserved[addr] = struct{}{} + 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 } - if !exists { - panic("not reserved") - } - delete(reserved, addr) - return nil - } } func setupPool() (*LegacyPool, *ecdsa.PrivateKey) { @@ -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) diff --git a/core/txpool/subpool.go b/core/txpool/subpool.go index f5cb852d8f..93e4763ae9 100644 --- a/core/txpool/subpool.go +++ b/core/txpool/subpool.go @@ -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. diff --git a/core/txpool/txpool.go b/core/txpool/txpool.go index 083aac92c6..1a8b684149 100644 --- a/core/txpool/txpool.go +++ b/core/txpool/txpool.go @@ -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 diff --git a/eth/backend.go b/eth/backend.go index 79759710b6..6716a77562 100644 --- a/eth/backend.go +++ b/eth/backend.go @@ -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 diff --git a/eth/protocols/eth/handler_test.go b/eth/protocols/eth/handler_test.go index bbbd888701..fa031d9899 100644 --- a/eth/protocols/eth/handler_test.go +++ b/eth/protocols/eth/handler_test.go @@ -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})