From 21b035eb29f6489ffee66d8ee2451873bc96dd2d Mon Sep 17 00:00:00 2001 From: Delweng Date: Mon, 7 Apr 2025 13:16:26 +0800 Subject: [PATCH 01/12] cmd/geth: set trie,gc and other cache flags for import chain (#31577) Signed-off-by: jsvisa --- cmd/geth/chaincmd.go | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/cmd/geth/chaincmd.go b/cmd/geth/chaincmd.go index 0c36b82af5..2279509542 100644 --- a/cmd/geth/chaincmd.go +++ b/cmd/geth/chaincmd.go @@ -79,11 +79,15 @@ if one is set. Otherwise it prints the genesis from the datadir.`, Usage: "Import a blockchain file", ArgsUsage: " ( ... ) ", Flags: slices.Concat([]cli.Flag{ - utils.CacheFlag, utils.GCModeFlag, utils.SnapshotFlag, + utils.CacheFlag, utils.CacheDatabaseFlag, + utils.CacheTrieFlag, utils.CacheGCFlag, + utils.CacheSnapshotFlag, + utils.CacheNoPrefetchFlag, + utils.CachePreimagesFlag, utils.NoCompactionFlag, utils.MetricsEnabledFlag, utils.MetricsEnabledExpensiveFlag, From ec6d1044045bcd6f8fe96892d840fe35c16ca7c8 Mon Sep 17 00:00:00 2001 From: Ocenka Date: Tue, 8 Apr 2025 15:44:13 +0300 Subject: [PATCH 02/12] eth/remotedb: improve error handling (#31331) This PR improves error handling in the remotedb package by fixing two issues: 1. In the `Has` method, we now properly propagate errors instead of silently returning false. This makes the behavior more predictable and helps clients better understand when there are connection issues. 2. In the `New` constructor, we add a nil check for the client parameter to prevent potential panics. This follows Go best practices for constructor functions. These changes make the code more robust and follow Go's error handling idioms without requiring any changes to other parts of the codebase. Changes: - Modified `Has` method to return errors instead of silently returning false - Added nil check in `New` constructor - Fixed field name in constructor to match struct definition --- ethdb/remotedb/remotedb.go | 9 +++++---- 1 file changed, 5 insertions(+), 4 deletions(-) diff --git a/ethdb/remotedb/remotedb.go b/ethdb/remotedb/remotedb.go index 247a4392db..8a91fdbcf2 100644 --- a/ethdb/remotedb/remotedb.go +++ b/ethdb/remotedb/remotedb.go @@ -34,7 +34,7 @@ type Database struct { func (db *Database) Has(key []byte) (bool, error) { if _, err := db.Get(key); err != nil { - return false, nil + return false, err } return true, nil } @@ -50,7 +50,7 @@ func (db *Database) Get(key []byte) ([]byte, error) { func (db *Database) HasAncient(kind string, number uint64) (bool, error) { if _, err := db.Ancient(kind, number); err != nil { - return false, nil + return false, err } return true, nil } @@ -144,7 +144,8 @@ func (db *Database) Close() error { } func New(client *rpc.Client) ethdb.Database { - return &Database{ - remote: client, + if client == nil { + return nil } + return &Database{remote: client} } From 2e739fce584a3cc8d84bf4c4604cd81c07294e2b Mon Sep 17 00:00:00 2001 From: rjl493456442 Date: Tue, 8 Apr 2025 21:46:27 +0800 Subject: [PATCH 03/12] core/txpool: add 7702 protection to blobpool (#31526) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit This pull request introduces two constraints in the blobPool: (a) If the sender has a pending authorization or delegation, only one in-flight executable transaction can be cached. (b) If the authority address in a SetCode transaction is already reserved by the blobPool, the transaction will be rejected. These constraints mitigate an attack where an attacker spams the pool with numerous blob transactions, evicts other transactions, and then cancels all pending blob transactions by draining the sender’s funds if they have a delegation. Note, because there is no exclusive lock held between different subpools when processing transactions, it's totally possible the SetCode transaction and blob transactions with conflict sender and authorities are accepted simultaneously. I think it's acceptable as it's very hard to be exploited. --------- Co-authored-by: lightclient --- core/txpool/blobpool/blobpool.go | 66 ++++++++--- core/txpool/blobpool/blobpool_test.go | 60 +++------- core/txpool/errors.go | 4 + core/txpool/legacypool/legacypool.go | 60 ++++++---- core/txpool/legacypool/legacypool2_test.go | 8 +- core/txpool/legacypool/legacypool_test.go | 82 +++++--------- core/txpool/reserver.go | 124 +++++++++++++++++++++ core/txpool/subpool.go | 6 +- core/txpool/txpool.go | 77 ++----------- eth/backend.go | 10 +- eth/protocols/eth/handler_test.go | 2 +- 11 files changed, 286 insertions(+), 213 deletions(-) create mode 100644 core/txpool/reserver.go diff --git a/core/txpool/blobpool/blobpool.go b/core/txpool/blobpool/blobpool.go index 5a20c3ce5a..b1966905a8 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 + 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 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,8 +355,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) error { - p.reserve = reserve +func (p *BlobPool) Init(gasTip uint64, head *types.Header, reserver *txpool.Reserver) error { + p.reserver = reserver var ( queuedir string @@ -499,7 +501,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); err != nil { return err } p.index[sender] = []*blobTxMeta{} @@ -554,7 +556,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) if gapped { log.Warn("Dropping dangling blob transactions", "from", addr, "missing", next, "drop", nonces, "ids", ids) @@ -707,7 +709,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) } else { p.index[addr] = txs } @@ -1006,7 +1008,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); err != nil { log.Warn("Failed to reserve account for blob pool", "tx", tx.Hash(), "from", addr, "err", err) return err } @@ -1066,7 +1068,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) } // 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) @@ -1101,6 +1103,39 @@ 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 + } + } + // Allow a single in-flight pending transaction. + pending := p.index[from] + if len(pending) == 0 { + return nil + } + // If account already has a pending transaction, allow replacement only. + 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 +1176,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 ( @@ -1369,7 +1407,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); err != nil { addNonExclusiveMeter.Mark(1) return err } @@ -1381,7 +1419,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) } }() } @@ -1513,7 +1551,7 @@ func (p *BlobPool) drop() { if last { delete(p.index, from) delete(p.spent, from) - p.reserve(from, false) + p.reserver.Release(from) } else { txs[len(txs)-1] = nil txs = txs[:len(txs)-1] @@ -1789,7 +1827,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) } 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 b7c6cfa51e..4dfba3b52b 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. @@ -433,6 +405,10 @@ func verifyBlobRetrievals(t *testing.T, pool *BlobPool) { } } +func newReserver() *txpool.Reserver { + return txpool.NewReservationTracker().NewHandle(42) +} + // Tests that transactions can be loaded from disk on startup and that they are // correctly discarded if invalid. // @@ -699,8 +675,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(), newReserver()); err != nil { t.Fatalf("failed to create blob pool: %v", err) } defer pool.Close() @@ -817,8 +793,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(), newReserver()); err != nil { t.Fatalf("failed to create blob pool: %v", err) } defer pool.Close() @@ -918,8 +894,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(), newReserver()); err != nil { t.Fatalf("failed to create blob pool: %v", err) } defer pool.Close() @@ -997,8 +973,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(), 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 @@ -1098,8 +1074,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(), newReserver()); err != nil { t.Fatalf("failed to create blob pool: %v", err) } @@ -1541,8 +1517,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(), newReserver()); err != nil { t.Fatalf("test %d: failed to create blob pool: %v", i, err) } verifyPoolInternals(t, pool) @@ -1638,10 +1614,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(), 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/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..278ad0791f 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") @@ -241,8 +237,8 @@ type LegacyPool struct { currentHead atomic.Pointer[types.Header] // Current head of the blockchain currentState *state.StateDB // Current state in the blockchain head pendingNonces *noncer // Pending state tracking virtual nonces + reserver *txpool.Reserver // Address reserver to ensure exclusivity across subpools - reserve txpool.AddressReserver // Address reserver to ensure exclusivity across subpools pending map[common.Address]*list // All currently processable transactions queue map[common.Address]*list // Queued but non-processable transactions beats map[common.Address]time.Time // Last heartbeat from each known account @@ -306,9 +302,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) 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.reserver = reserver // Set the basic pool parameters pool.gasTip.Store(uint256.NewInt(gasTip)) @@ -618,7 +614,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 +629,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 +640,24 @@ func (pool *LegacyPool) validateAuth(tx *types.Transaction) error { if err := pool.checkDelegationLimit(tx); err != nil { return err } - // Authorities cannot conflict with any pending or queued transactions. + // Authorities must not conflict with any pending or queued transactions, + // nor with addresses that have already been reserved. if auths := tx.SetCodeAuthorities(); len(auths) > 0 { for _, auth := range auths { if pool.pending[auth] != nil || pool.queue[auth] != nil { return ErrAuthorityReserved } + // Because there is no exclusive lock held between different subpools + // when processing transactions, the SetCode transaction may be accepted + // while other transactions with the same sender address are also + // accepted simultaneously in the other pools. + // + // This scenario is considered acceptable, as the rule primarily ensures + // that attackers cannot easily stack a SetCode transaction when the sender + // is reserved by other pools. + if pool.reserver.Has(auth) { + return ErrAuthorityReserved + } } } return nil @@ -683,7 +691,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); err != nil { return false, err } defer func() { @@ -694,7 +702,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) } }() } @@ -1087,7 +1095,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) } }() } @@ -1467,7 +1475,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) } } } @@ -1653,7 +1661,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) } } } @@ -1862,11 +1870,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 +1890,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,11 +1902,11 @@ 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) - pool.reserve(senderAddr, false) + pool.reserver.Release(senderAddr) } pool.all = newLookup() pool.priced = newPricedList(pool.all) @@ -1904,3 +1914,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..3f210e3d1b 100644 --- a/core/txpool/legacypool/legacypool2_test.go +++ b/core/txpool/legacypool/legacypool2_test.go @@ -86,7 +86,7 @@ func TestTransactionFutureAttack(t *testing.T) { config.GlobalQueue = 100 config.GlobalSlots = 100 pool := New(config, blockchain) - pool.Init(config.PriceLimit, blockchain.CurrentBlock(), makeAddressReserver()) + pool.Init(config.PriceLimit, blockchain.CurrentBlock(), newReserver()) defer pool.Close() fillPool(t, pool) pending, _ := pool.Stats() @@ -120,7 +120,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) - pool.Init(testTxPoolConfig.PriceLimit, blockchain.CurrentBlock(), makeAddressReserver()) + pool.Init(testTxPoolConfig.PriceLimit, blockchain.CurrentBlock(), newReserver()) defer pool.Close() // Create a number of test accounts, fund them and make transactions @@ -153,7 +153,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) - pool.Init(testTxPoolConfig.PriceLimit, blockchain.CurrentBlock(), makeAddressReserver()) + pool.Init(testTxPoolConfig.PriceLimit, blockchain.CurrentBlock(), newReserver()) defer pool.Close() // Create a number of test accounts, fund them and make transactions fillPool(t, pool) @@ -224,7 +224,7 @@ func BenchmarkFutureAttack(b *testing.B) { config.GlobalQueue = 100 config.GlobalSlots = 100 pool := New(config, blockchain) - pool.Init(testTxPoolConfig.PriceLimit, blockchain.CurrentBlock(), makeAddressReserver()) + pool.Init(testTxPoolConfig.PriceLimit, blockchain.CurrentBlock(), newReserver()) defer pool.Close() fillPool(b, pool) diff --git a/core/txpool/legacypool/legacypool_test.go b/core/txpool/legacypool/legacypool_test.go index 3f269bd69e..c47a655204 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,42 +167,21 @@ func pricedSetCodeTxWithAuth(nonce uint64, gaslimit uint64, gasFee, tip *uint256 }) } -func makeAddressReserver() txpool.AddressReserver { - var ( - reserved = make(map[common.Address]struct{}) - lock sync.Mutex - ) - return func(addr common.Address, reserve bool) error { - lock.Lock() - defer lock.Unlock() - - _, exists := reserved[addr] - if reserve { - if exists { - panic("already reserved") - } - reserved[addr] = struct{}{} - return nil - } - if !exists { - panic("not reserved") - } - delete(reserved, addr) - return nil - } -} - func setupPool() (*LegacyPool, *ecdsa.PrivateKey) { return setupPoolWithConfig(params.TestChainConfig) } +func newReserver() *txpool.Reserver { + return txpool.NewReservationTracker().NewHandle(42) +} + func setupPoolWithConfig(config *params.ChainConfig) (*LegacyPool, *ecdsa.PrivateKey) { statedb, _ := state.New(types.EmptyRootHash, state.NewDatabaseForTesting()) blockchain := newTestBlockChain(config, 10000000, statedb, new(event.Feed)) key, _ := crypto.GenerateKey() pool := New(testTxPoolConfig, blockchain) - if err := pool.Init(testTxPoolConfig.PriceLimit, blockchain.CurrentBlock(), makeAddressReserver()); err != nil { + if err := pool.Init(testTxPoolConfig.PriceLimit, blockchain.CurrentBlock(), newReserver()); err != nil { panic(err) } // wait for the pool to initialize @@ -336,7 +314,7 @@ func TestStateChangeDuringReset(t *testing.T) { tx1 := transaction(1, 100000, key) pool := New(testTxPoolConfig, blockchain) - pool.Init(testTxPoolConfig.PriceLimit, blockchain.CurrentBlock(), makeAddressReserver()) + pool.Init(testTxPoolConfig.PriceLimit, blockchain.CurrentBlock(), newReserver()) defer pool.Close() nonce := pool.Nonce(address) @@ -753,7 +731,7 @@ 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()) + pool.Init(testTxPoolConfig.PriceLimit, blockchain.CurrentBlock(), newReserver()) defer pool.Close() // Create two test accounts to produce different gap profiles with @@ -963,7 +941,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) - pool.Init(testTxPoolConfig.PriceLimit, blockchain.CurrentBlock(), makeAddressReserver()) + pool.Init(testTxPoolConfig.PriceLimit, blockchain.CurrentBlock(), newReserver()) defer pool.Close() // Create a number of test accounts and fund them (last one will be the local) @@ -1015,7 +993,7 @@ func TestQueueTimeLimiting(t *testing.T) { config.Lifetime = time.Second pool := New(config, blockchain) - pool.Init(config.PriceLimit, blockchain.CurrentBlock(), makeAddressReserver()) + pool.Init(config.PriceLimit, blockchain.CurrentBlock(), newReserver()) defer pool.Close() // Create a test account to ensure remotes expire @@ -1176,7 +1154,7 @@ func TestPendingGlobalLimiting(t *testing.T) { config.GlobalSlots = config.AccountSlots * 10 pool := New(config, blockchain) - pool.Init(config.PriceLimit, blockchain.CurrentBlock(), makeAddressReserver()) + pool.Init(config.PriceLimit, blockchain.CurrentBlock(), newReserver()) defer pool.Close() // Create a number of test accounts and fund them @@ -1275,7 +1253,7 @@ func TestCapClearsFromAll(t *testing.T) { config.GlobalSlots = 8 pool := New(config, blockchain) - pool.Init(config.PriceLimit, blockchain.CurrentBlock(), makeAddressReserver()) + pool.Init(config.PriceLimit, blockchain.CurrentBlock(), newReserver()) defer pool.Close() // Create a number of test accounts and fund them @@ -1308,7 +1286,7 @@ func TestPendingMinimumAllowance(t *testing.T) { config.GlobalSlots = 1 pool := New(config, blockchain) - pool.Init(config.PriceLimit, blockchain.CurrentBlock(), makeAddressReserver()) + pool.Init(config.PriceLimit, blockchain.CurrentBlock(), newReserver()) defer pool.Close() // Create a number of test accounts and fund them @@ -1352,7 +1330,7 @@ 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()) + pool.Init(testTxPoolConfig.PriceLimit, blockchain.CurrentBlock(), newReserver()) defer pool.Close() // Keep track of transaction events to ensure all executables get announced @@ -1457,7 +1435,7 @@ func TestMinGasPriceEnforced(t *testing.T) { txPoolConfig := DefaultConfig txPoolConfig.NoLocals = true pool := New(txPoolConfig, blockchain) - pool.Init(txPoolConfig.PriceLimit, blockchain.CurrentBlock(), makeAddressReserver()) + pool.Init(txPoolConfig.PriceLimit, blockchain.CurrentBlock(), newReserver()) defer pool.Close() key, _ := crypto.GenerateKey() @@ -1606,7 +1584,7 @@ func TestUnderpricing(t *testing.T) { config.GlobalQueue = 2 pool := New(config, blockchain) - pool.Init(config.PriceLimit, blockchain.CurrentBlock(), makeAddressReserver()) + pool.Init(config.PriceLimit, blockchain.CurrentBlock(), newReserver()) defer pool.Close() // Keep track of transaction events to ensure all executables get announced @@ -1696,7 +1674,7 @@ func TestStableUnderpricing(t *testing.T) { config.GlobalQueue = 0 pool := New(config, blockchain) - pool.Init(config.PriceLimit, blockchain.CurrentBlock(), makeAddressReserver()) + pool.Init(config.PriceLimit, blockchain.CurrentBlock(), newReserver()) defer pool.Close() // Keep track of transaction events to ensure all executables get announced @@ -1899,7 +1877,7 @@ 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()) + pool.Init(testTxPoolConfig.PriceLimit, blockchain.CurrentBlock(), newReserver()) defer pool.Close() // Create a test account to add transactions with @@ -1966,7 +1944,7 @@ 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()) + pool.Init(testTxPoolConfig.PriceLimit, blockchain.CurrentBlock(), newReserver()) defer pool.Close() // Keep track of transaction events to ensure all executables get announced @@ -2157,7 +2135,7 @@ 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()) + pool.Init(testTxPoolConfig.PriceLimit, blockchain.CurrentBlock(), newReserver()) defer pool.Close() // Create the test accounts to check various transaction statuses with @@ -2230,7 +2208,7 @@ 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()) + pool.Init(testTxPoolConfig.PriceLimit, blockchain.CurrentBlock(), newReserver()) defer pool.Close() // Create the test accounts @@ -2271,12 +2249,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 +2288,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 +2364,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 +2432,7 @@ 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()) + pool.Init(testTxPoolConfig.PriceLimit, blockchain.CurrentBlock(), newReserver()) defer pool.Close() // Create the test accounts @@ -2489,8 +2467,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/reserver.go b/core/txpool/reserver.go new file mode 100644 index 0000000000..76ead0f3bb --- /dev/null +++ b/core/txpool/reserver.go @@ -0,0 +1,124 @@ +// Copyright 2025 The go-ethereum Authors +// This file is part of the go-ethereum library. +// +// The go-ethereum library is free software: you can redistribute it and/or modify +// it under the terms of the GNU Lesser General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. +// +// The go-ethereum library is distributed in the hope that it will be useful, +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU Lesser General Public License for more details. +// +// You should have received a copy of the GNU Lesser General Public License +// along with the go-ethereum library. If not, see . + +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" +) + +// ReservationTracker is a struct shared between different subpools. It is used to reserve +// the account and ensure that one address cannot initiate transactions, authorizations, +// and other state-changing behaviors in different pools at the same time. +type ReservationTracker struct { + accounts map[common.Address]int + lock sync.RWMutex +} + +// NewReservationTracker initializes the account reservation tracker. +func NewReservationTracker() *ReservationTracker { + return &ReservationTracker{ + accounts: make(map[common.Address]int), + } +} + +// NewHandle creates a named handle on the ReservationTracker. The handle +// identifies the subpool so ownership of reservations can be determined. +func (r *ReservationTracker) NewHandle(id int) *Reserver { + return &Reserver{r, id} +} + +// Reserver is a named handle on ReservationTracker. It is held by subpools to +// make reservations for accounts it is tracking. The id is used to determine +// which pool owns an address and disallows non-owners to hold or release +// addresses it doesn't own. +type Reserver struct { + tracker *ReservationTracker + id int +} + +// Hold attempts to reserve the specified account address for the given pool. +// Returns an error if the account is already reserved. +func (h *Reserver) Hold(addr common.Address) error { + h.tracker.lock.Lock() + defer h.tracker.lock.Unlock() + + // Double reservations are forbidden even from the same pool to + // avoid subtle bugs in the long term. + owner, exists := h.tracker.accounts[addr] + if exists { + if owner == h.id { + log.Error("pool attempted to reserve already-owned address", "address", addr) + return nil // Ignore fault to give the pool a chance to recover while the bug gets fixed + } + return ErrAlreadyReserved + } + h.tracker.accounts[addr] = h.id + if metrics.Enabled() { + m := fmt.Sprintf("%s/%d", reservationsGaugeName, h.id) + metrics.GetOrRegisterGauge(m, nil).Inc(1) + } + return nil +} + +// Release attempts to release the reservation for the specified account. +// Returns an error if the address is not reserved or is reserved by another pool. +func (h *Reserver) Release(addr common.Address) error { + h.tracker.lock.Lock() + defer h.tracker.lock.Unlock() + + // Ensure subpools only attempt to unreserve their own owned addresses, + // otherwise flag as a programming error. + owner, exists := h.tracker.accounts[addr] + if !exists { + log.Error("pool attempted to unreserve non-reserved address", "address", addr) + return errors.New("address not reserved") + } + if owner != h.id { + log.Error("pool attempted to unreserve non-owned address", "address", addr) + return errors.New("address not owned") + } + delete(h.tracker.accounts, addr) + if metrics.Enabled() { + m := fmt.Sprintf("%s/%d", reservationsGaugeName, h.id) + metrics.GetOrRegisterGauge(m, nil).Dec(1) + } + return nil +} + +// Has returns a flag indicating if the address has been reserved or not. +func (h *Reserver) Has(address common.Address) bool { + h.tracker.lock.RLock() + defer h.tracker.lock.RUnlock() + + _, exists := h.tracker.accounts[address] + return exists +} diff --git a/core/txpool/subpool.go b/core/txpool/subpool.go index f5cb852d8f..2cb5103875 100644 --- a/core/txpool/subpool.go +++ b/core/txpool/subpool.go @@ -67,10 +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 - // PendingFilter is a collection of filter rules to allow retrieving a subset // of transactions for announcement or mining. // @@ -109,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) 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 083aac92c6..47d83e03d4 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.Mutex // Lock protecting the account reservations - subs event.SubscriptionScope // Subscription scope to unsubscribe all on shutdown quit chan chan error // Quit channel to tear down the head updater 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 := NewReservationTracker() for i, subpool := range subpools { - if err := subpool.Init(gasTip, head, pool.reserver(i, subpool)); err != nil { + if err := subpool.Init(gasTip, head, reserver.NewHandle(i)); err != nil { for j := i - 1; j >= 0; j-- { subpools[j].Close() } @@ -131,52 +118,6 @@ func New(gasTip uint64, chain BlockChain, subpools []SubPool) (*TxPool, error) { return pool, nil } -// reserver is a method to create an address reservation callback to exclusively -// assign/deassign addresses to/from subpools. This can ensure that at any point -// in time, only a single subpool is able to manage an account, avoiding cross -// subpool eviction issues and nonce conflicts. -func (p *TxPool) reserver(id int, subpool SubPool) AddressReserver { - return func(addr common.Address, reserve bool) error { - p.reserveLock.Lock() - defer p.reserveLock.Unlock() - - owner, exists := p.reservations[addr] - if reserve { - // Double reservations are forbidden even from the same pool to - // avoid subtle bugs in the long term. - if exists { - if owner == subpool { - log.Error("pool attempted to reserve already-owned address", "address", addr) - return nil // Ignore fault to give the pool a chance to recover while the bug gets fixed - } - return ErrAlreadyReserved - } - p.reservations[addr] = subpool - if metrics.Enabled() { - m := fmt.Sprintf("%s/%d", reservationsGaugeName, id) - metrics.GetOrRegisterGauge(m, nil).Inc(1) - } - return nil - } - // Ensure subpools only attempt to unreserve their own owned addresses, - // otherwise flag as a programming error. - if !exists { - log.Error("pool attempted to unreserve non-reserved address", "address", addr) - return errors.New("address not reserved") - } - if subpool != owner { - log.Error("pool attempted to unreserve non-owned address", "address", addr) - return errors.New("address not owned") - } - delete(p.reservations, addr) - if metrics.Enabled() { - m := fmt.Sprintf("%s/%d", reservationsGaugeName, id) - metrics.GetOrRegisterGauge(m, nil).Dec(1) - } - return nil - } -} - // Close terminates the transaction pool and all its subpools. func (p *TxPool) Close() error { var errs []error 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}) From 5cc9137c9cec8a977b845a025d3deced9457fd48 Mon Sep 17 00:00:00 2001 From: Marius van der Wijden Date: Tue, 8 Apr 2025 19:57:45 +0200 Subject: [PATCH 04/12] core/vm: optimize push2 opcode (#31267) During my benchmarks on Holesky, around 10% of all CPU time was spent in PUSH2 ``` ROUTINE ======================== github.com/ethereum/go-ethereum/core/vm.newFrontierInstructionSet.makePush.func1 in github.com/ethereum/go-ethereum/core/vm/instructions.go 16.38s 20.35s (flat, cum) 10.31% of Total 740ms 740ms 976: return func(pc *uint64, interpreter *EVMInterpreter, scope *ScopeContext) ([]byte, error) { . . 977: var ( 40ms 40ms 978: codeLen = len(scope.Contract.Code) 970ms 970ms 979: start = min(codeLen, int(*pc+1)) 200ms 200ms 980: end = min(codeLen, start+pushByteSize) . . 981: ) 670ms 2.39s 982: a := new(uint256.Int).SetBytes(scope.Contract.Code[start:end]) . . 983: . . 984: // Missing bytes: pushByteSize - len(pushData) 410ms 410ms 985: if missing := pushByteSize - (end - start); missing > 0 { . . 986: a.Lsh(a, uint(8*missing)) . . 987: } 12.69s 14.94s 988: scope.Stack.push2(*a) 10ms 10ms 989: *pc += size 650ms 650ms 990: return nil, nil . . 991: } . . 992:} ``` Which is quite crazy. We have a handwritten encoder for PUSH1 already, this PR adds one for PUSH2. PUSH2 is the second most used opcode as shown here: https://gist.github.com/shemnon/fb9b292a103abb02d98d64df6fbd35c8 since it is used by solidity quite significantly. Its used ~20 times as much as PUSH20 and PUSH32. # Benchmarks ``` BenchmarkPush/makePush-14 94196547 12.27 ns/op 0 B/op 0 allocs/op BenchmarkPush/push-14 429976924 2.829 ns/op 0 B/op 0 allocs/op ``` --------- Co-authored-by: jwasinger --- core/vm/instructions.go | 17 +++++++++++++++++ core/vm/jump_table.go | 2 +- 2 files changed, 18 insertions(+), 1 deletion(-) diff --git a/core/vm/instructions.go b/core/vm/instructions.go index 1785ffc139..0b3b1d1569 100644 --- a/core/vm/instructions.go +++ b/core/vm/instructions.go @@ -971,6 +971,23 @@ func opPush1(pc *uint64, interpreter *EVMInterpreter, scope *ScopeContext) ([]by return nil, nil } +// opPush2 is a specialized version of pushN +func opPush2(pc *uint64, interpreter *EVMInterpreter, scope *ScopeContext) ([]byte, error) { + var ( + codeLen = uint64(len(scope.Contract.Code)) + integer = new(uint256.Int) + ) + if *pc+2 < codeLen { + scope.Stack.push(integer.SetBytes2(scope.Contract.Code[*pc+1 : *pc+3])) + } else if *pc+1 < codeLen { + scope.Stack.push(integer.SetUint64(uint64(scope.Contract.Code[*pc+1]) << 8)) + } else { + scope.Stack.push(integer.Clear()) + } + *pc += 2 + return nil, nil +} + // make push instruction function func makePush(size uint64, pushByteSize int) executionFunc { return func(pc *uint64, interpreter *EVMInterpreter, scope *ScopeContext) ([]byte, error) { diff --git a/core/vm/jump_table.go b/core/vm/jump_table.go index 6610fa7f9a..ee811b447e 100644 --- a/core/vm/jump_table.go +++ b/core/vm/jump_table.go @@ -631,7 +631,7 @@ func newFrontierInstructionSet() JumpTable { maxStack: maxStack(0, 1), }, PUSH2: { - execute: makePush(2, 2), + execute: opPush2, constantGas: GasFastestStep, minStack: minStack(0, 1), maxStack: maxStack(0, 1), From a7f24c26c0a6a73357d76adf64d98f343bafd0e5 Mon Sep 17 00:00:00 2001 From: Csaba Kiraly Date: Wed, 9 Apr 2025 11:28:29 +0200 Subject: [PATCH 05/12] p2p/nat: fix UPnP port reset (#31566) Make UPnP more robust - Once a random port was mapped, we try to stick to it even if a UPnP refresh fails. Previously we were immediately moving back to try the default port, leading to frequent ENR changes. - We were deleting port mappings before refresh as a possible workaround. This created issues in some UPnP servers. The UPnP (and PMP) specification is explicit about the refresh requirements, and delete is clearly not needed (see https://github.com/ethereum/go-ethereum/pull/30265#issuecomment-2766987859). From now on we only delete when closing. - We were trying to add port mappings only once, and then moved on to random ports. Now we insist a bit more, so that a simple failed request won't lead to ENR changes. Fixes https://github.com/ethereum/go-ethereum/issues/31418 --------- Signed-off-by: Csaba Kiraly Co-authored-by: Felix Lange --- p2p/nat/natupnp.go | 49 +++++++++++++++++++++++++-------------- p2p/server_nat.go | 57 ++++++++++++++++++++++++++++++++-------------- 2 files changed, 72 insertions(+), 34 deletions(-) diff --git a/p2p/nat/natupnp.go b/p2p/nat/natupnp.go index ed00b8eeb6..d79677db55 100644 --- a/p2p/nat/natupnp.go +++ b/p2p/nat/natupnp.go @@ -26,6 +26,7 @@ import ( "sync" "time" + "github.com/ethereum/go-ethereum/log" "github.com/huin/goupnp" "github.com/huin/goupnp/dcps/internetgateway1" "github.com/huin/goupnp/dcps/internetgateway2" @@ -34,6 +35,8 @@ import ( const ( soapRequestTimeout = 3 * time.Second rateLimit = 200 * time.Millisecond + retryCount = 3 // number of retries after a failed AddPortMapping + randomCount = 3 // number of random ports to try ) type upnp struct { @@ -89,42 +92,43 @@ func (n *upnp) AddMapping(protocol string, extport, intport int, desc string, li if extport == 0 { extport = intport - } else { - // Only delete port mapping if the external port was already used by geth. - n.DeleteMapping(protocol, extport, intport) } // Try to add port mapping, preferring the specified external port. - err = n.withRateLimit(func() error { - p, err := n.addAnyPortMapping(protocol, extport, intport, ip, desc, lifetimeS) - if err == nil { - extport = int(p) - } - return err - }) - return uint16(extport), err + return n.addAnyPortMapping(protocol, extport, intport, ip, desc, lifetimeS) } // addAnyPortMapping tries to add a port mapping with the specified external port. // If the external port is already in use, it will try to assign another port. func (n *upnp) addAnyPortMapping(protocol string, extport, intport int, ip net.IP, desc string, lifetimeS uint32) (uint16, error) { if client, ok := n.client.(*internetgateway2.WANIPConnection2); ok { - return client.AddAnyPortMapping("", uint16(extport), protocol, uint16(intport), ip.String(), true, desc, lifetimeS) + return n.portWithRateLimit(func() (uint16, error) { + return client.AddAnyPortMapping("", uint16(extport), protocol, uint16(intport), ip.String(), true, desc, lifetimeS) + }) } // For IGDv1 and v1 services we should first try to add with extport. - err := n.client.AddPortMapping("", uint16(extport), protocol, uint16(intport), ip.String(), true, desc, lifetimeS) - if err == nil { - return uint16(extport), nil + for i := 0; i < retryCount+1; i++ { + err := n.withRateLimit(func() error { + return n.client.AddPortMapping("", uint16(extport), protocol, uint16(intport), ip.String(), true, desc, lifetimeS) + }) + if err == nil { + return uint16(extport), nil + } + log.Debug("Failed to add port mapping", "protocol", protocol, "extport", extport, "intport", intport, "err", err) } // If above fails, we retry with a random port. // We retry several times because of possible port conflicts. - for i := 0; i < 3; i++ { + var err error + for i := 0; i < randomCount; i++ { extport = n.randomPort() - err := n.client.AddPortMapping("", uint16(extport), protocol, uint16(intport), ip.String(), true, desc, lifetimeS) + err := n.withRateLimit(func() error { + return n.client.AddPortMapping("", uint16(extport), protocol, uint16(intport), ip.String(), true, desc, lifetimeS) + }) if err == nil { return uint16(extport), nil } + log.Debug("Failed to add random port mapping", "protocol", protocol, "extport", extport, "intport", intport, "err", err) } return 0, err } @@ -169,6 +173,17 @@ func (n *upnp) String() string { return "UPNP " + n.service } +func (n *upnp) portWithRateLimit(pfn func() (uint16, error)) (uint16, error) { + var port uint16 + var err error + fn := func() error { + port, err = pfn() + return err + } + n.withRateLimit(fn) + return port, err +} + func (n *upnp) withRateLimit(fn func() error) error { n.mu.Lock() defer n.mu.Unlock() diff --git a/p2p/server_nat.go b/p2p/server_nat.go index 5830f950e1..298c454a4a 100644 --- a/p2p/server_nat.go +++ b/p2p/server_nat.go @@ -31,12 +31,14 @@ const ( portMapRefreshInterval = 8 * time.Minute portMapRetryInterval = 5 * time.Minute extipRetryInterval = 2 * time.Minute + maxRetries = 5 // max number of failed attempts to refresh the mapping ) type portMapping struct { protocol string name string port int + retries int // number of failed attempts to refresh the mapping // for use by the portMappingLoop goroutine: extPort int // the mapped port returned by the NAT interface @@ -154,28 +156,49 @@ func (srv *Server) portMappingLoop() { log.Trace("Attempting port mapping") p, err := srv.NAT.AddMapping(m.protocol, m.extPort, m.port, m.name, portMapDuration) if err != nil { - log.Debug("Couldn't add port mapping", "err", err) - m.extPort = 0 + // Failed to add or refresh port mapping. + if m.extPort == 0 { + log.Debug("Couldn't add port mapping", "err", err) + } else { + // Failed refresh. Since UPnP implementation are often buggy, + // and lifetime is larger than the retry interval, this does not + // mean we lost our existing mapping. We do not reset the external + // port, as it is still our best chance, but we do retry soon. + // We could check the error code, but UPnP implementations are buggy. + log.Debug("Couldn't refresh port mapping", "err", err) + m.retries++ + if m.retries > maxRetries { + m.retries = 0 + err := srv.NAT.DeleteMapping(m.protocol, m.extPort, m.port) + log.Debug("Couldn't refresh port mapping, trying to delete it:", "err", err) + m.extPort = 0 + } + } m.nextTime = srv.clock.Now().Add(portMapRetryInterval) + // Note ENR is not updated here, i.e. we keep the last port. continue } - // It was mapped! - m.extPort = int(p) - m.nextTime = srv.clock.Now().Add(portMapRefreshInterval) - log = newLogger(m.protocol, m.extPort, m.port) - if m.port != m.extPort { - log.Info("NAT mapped alternative port") - } else { - log.Info("NAT mapped port") - } - // Update port in local ENR. - switch m.protocol { - case "TCP": - srv.localnode.Set(enr.TCP(m.extPort)) - case "UDP": - srv.localnode.SetFallbackUDP(m.extPort) + // It was mapped! + m.retries = 0 + log = newLogger(m.protocol, int(p), m.port) + if int(p) != m.extPort { + m.extPort = int(p) + if m.port != m.extPort { + log.Info("NAT mapped alternative port") + } else { + log.Info("NAT mapped port") + } + + // Update port in local ENR. + switch m.protocol { + case "TCP": + srv.localnode.Set(enr.TCP(m.extPort)) + case "UDP": + srv.localnode.SetFallbackUDP(m.extPort) + } } + m.nextTime = srv.clock.Now().Add(portMapRefreshInterval) } } } From 60b922fd529237ac5070a1750c93a4c132ed32fb Mon Sep 17 00:00:00 2001 From: lightclient <14004106+lightclient@users.noreply.github.com> Date: Wed, 9 Apr 2025 14:19:28 -0600 Subject: [PATCH 06/12] core/txpool: add notice to` Clear` that is not meant for production code (#31567) The `Sync(..)` function explicitly says not to rely on in production code, but it is used in `Clear(..)` so should add a similar mention. --- core/txpool/blobpool/blobpool.go | 6 ++++++ core/txpool/legacypool/legacypool.go | 7 +++++-- core/txpool/txpool.go | 11 +++++++++-- 3 files changed, 20 insertions(+), 4 deletions(-) diff --git a/core/txpool/blobpool/blobpool.go b/core/txpool/blobpool/blobpool.go index b1966905a8..1770066a8d 100644 --- a/core/txpool/blobpool/blobpool.go +++ b/core/txpool/blobpool/blobpool.go @@ -1349,6 +1349,9 @@ func (p *BlobPool) GetBlobs(vhashes []common.Hash) ([]*kzg4844.Blob, []*kzg4844. // Add inserts a set of blob transactions into the pool if they pass validation (both // consensus validity and pool restrictions). +// +// Note, if sync is set the method will block until all internal maintenance +// related to the add is finished. Only use this during tests for determinism. func (p *BlobPool) Add(txs []*types.Transaction, sync bool) []error { var ( adds = make([]*types.Transaction, 0, len(txs)) @@ -1792,6 +1795,9 @@ func (p *BlobPool) Status(hash common.Hash) txpool.TxStatus { // Clear implements txpool.SubPool, removing all tracked transactions // from the blob pool and persistent store. +// +// Note, do not use this in production / live code. In live code, the pool is +// meant to reset on a separate thread to avoid DoS vectors. func (p *BlobPool) Clear() { p.lock.Lock() defer p.lock.Unlock() diff --git a/core/txpool/legacypool/legacypool.go b/core/txpool/legacypool/legacypool.go index 278ad0791f..75dc4a8461 100644 --- a/core/txpool/legacypool/legacypool.go +++ b/core/txpool/legacypool/legacypool.go @@ -927,8 +927,8 @@ func (pool *LegacyPool) addRemoteSync(tx *types.Transaction) error { // Add enqueues a batch of transactions into the pool if they are valid. // -// If sync is set, the method will block until all internal maintenance related -// to the add is finished. Only use this during tests for determinism! +// Note, if sync is set the method will block until all internal maintenance +// related to the add is finished. Only use this during tests for determinism. func (pool *LegacyPool) Add(txs []*types.Transaction, sync bool) []error { // Filter out known ones without obtaining the pool lock or recovering signatures var ( @@ -1886,6 +1886,9 @@ func numSlots(tx *types.Transaction) int { // Clear implements txpool.SubPool, removing all tracked txs from the pool // and rotating the journal. +// +// Note, do not use this in production / live code. In live code, the pool is +// meant to reset on a separate thread to avoid DoS vectors. func (pool *LegacyPool) Clear() { pool.mu.Lock() defer pool.mu.Unlock() diff --git a/core/txpool/txpool.go b/core/txpool/txpool.go index 47d83e03d4..2ed38772ce 100644 --- a/core/txpool/txpool.go +++ b/core/txpool/txpool.go @@ -350,6 +350,9 @@ func (p *TxPool) ValidateTxBasics(tx *types.Transaction) error { // Add enqueues a batch of transactions into the pool if they are valid. Due // to the large transaction churn, add may postpone fully integrating the tx // to a later point to batch multiple ones together. +// +// Note, if sync is set the method will block until all internal maintenance +// related to the add is finished. Only use this during tests for determinism. func (p *TxPool) Add(txs []*types.Transaction, sync bool) []error { // Split the input transactions between the subpools. It shouldn't really // happen that we receive merged batches, but better graceful than strange @@ -503,8 +506,8 @@ func (p *TxPool) Status(hash common.Hash) TxStatus { // internal background reset operations. This method will run an explicit reset // operation to ensure the pool stabilises, thus avoiding flakey behavior. // -// Note, do not use this in production / live code. In live code, the pool is -// meant to reset on a separate thread to avoid DoS vectors. +// Note, this method is only used for testing and is susceptible to DoS vectors. +// In production code, the pool is meant to reset on a separate thread. func (p *TxPool) Sync() error { sync := make(chan error) select { @@ -516,6 +519,10 @@ func (p *TxPool) Sync() error { } // Clear removes all tracked txs from the subpools. +// +// Note, this method invokes Sync() and is only used for testing, because it is +// susceptible to DoS vectors. In production code, the pool is meant to reset on +// a separate thread. func (p *TxPool) Clear() { // Invoke Sync to ensure that txs pending addition don't get added to the pool after // the subpools are subsequently cleared From 0c2ad076734e87fcc008e4718ba7b1f4f881b1fa Mon Sep 17 00:00:00 2001 From: Luis Ayuso Date: Thu, 10 Apr 2025 03:11:24 +0200 Subject: [PATCH 07/12] core/txpool: allow tx and authority regardless of admission order (#31373) This PR proposes a change to the authorizations' validation introduced in commit cdb66c8. These changes make the expected behavior independent of the order of admission of authorizations, improving the predictability of the resulting state and the usability of the system with it. The current implementation behavior is dependent on the transaction submission order: This issue is related to authorities and the sender of a transaction, and can be reproduced respecting the normal nonce rules. The issue can be reproduced by the two following cases: **First case** - Given an empty pool. - Submit transaction `{ from: B, auths [ A ] }`: is accepted. - Submit transaction `{ from: A }`: Is accepted: it becomes the one in-flight transaction allowed. **Second case** - Given an empty pool. - Submit transaction `{ from: A }`: is accepted - Submit transaction `{ from: B, auths [ A ] }`: is rejected since there is already a queued/pending transaction from A. The expected behavior is that both sequences of events would lead to the same sets of accepted and rejected transactions. **Proposed changes** The queued/pending transactions issued from any authority of the transaction being validated have to be counted, allowing one transaction from accounts submitting an authorization. - Notice that the expected behavior was explicitly forbidden in the case "reject-delegation-from-pending-account", I believe that this behavior conflicts to the definition of the limitation, and it is removed in this PR. The expected behavior is tested in "accept-authorization-from-sender-of-one-inflight-tx". - Replacement tests have been separated to improve readability of the acceptance test. - The test "allow-more-than-one-tx-from-replaced-authority" has been extended with one extra transaction, since the system would always have accepted one transaction (but not two). - The test "accept-one-inflight-tx-of-delegated-account" is extended to clean-up state, avoiding leaking the delegation used into the other tests. Additionally, replacement check is removed to be tested in its own test case. **Expected behavior** The expected behavior of the authorizations' validation shall be as follows: ![image](https://github.com/user-attachments/assets/dbde7a1f-9679-4691-94eb-c197a0cbb823) Notice that replacement shall be allowed, and behavior shall remain coherent with the table, according to the replaced transaction. --------- Co-authored-by: lightclient --- core/txpool/blobpool/blobpool.go | 4 +- core/txpool/blobpool/blobpool_test.go | 43 ++++- core/txpool/legacypool/legacypool.go | 28 ++- core/txpool/legacypool/legacypool_test.go | 198 +++++++++++++++++----- core/txpool/reserver.go | 42 +++-- core/txpool/subpool.go | 2 +- 6 files changed, 242 insertions(+), 75 deletions(-) diff --git a/core/txpool/blobpool/blobpool.go b/core/txpool/blobpool/blobpool.go index 1770066a8d..12a4133b40 100644 --- a/core/txpool/blobpool/blobpool.go +++ b/core/txpool/blobpool/blobpool.go @@ -299,7 +299,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 - reserver *txpool.Reserver // 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,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, reserver *txpool.Reserver) error { +func (p *BlobPool) Init(gasTip uint64, head *types.Header, reserver txpool.Reserver) error { p.reserver = reserver var ( diff --git a/core/txpool/blobpool/blobpool_test.go b/core/txpool/blobpool/blobpool_test.go index 4dfba3b52b..76d21a0c9e 100644 --- a/core/txpool/blobpool/blobpool_test.go +++ b/core/txpool/blobpool/blobpool_test.go @@ -26,6 +26,7 @@ import ( "math/big" "os" "path/filepath" + "sync" "testing" "github.com/ethereum/go-ethereum/common" @@ -167,6 +168,44 @@ func (bc *testBlockChain) StateAt(common.Hash) (*state.StateDB, error) { return bc.statedb, nil } +// reserver is a utility struct to sanity check that accounts are +// properly reserved by the blobpool (no duplicate reserves or unreserves). +type reserver struct { + accounts map[common.Address]struct{} + lock sync.RWMutex +} + +func newReserver() txpool.Reserver { + return &reserver{accounts: make(map[common.Address]struct{})} +} + +func (r *reserver) Hold(addr common.Address) error { + r.lock.Lock() + defer r.lock.Unlock() + if _, exists := r.accounts[addr]; exists { + panic("already reserved") + } + r.accounts[addr] = struct{}{} + return nil +} + +func (r *reserver) Release(addr common.Address) error { + r.lock.Lock() + defer r.lock.Unlock() + if _, exists := r.accounts[addr]; !exists { + panic("not reserved") + } + delete(r.accounts, addr) + return nil +} + +func (r *reserver) Has(address common.Address) bool { + r.lock.RLock() + defer r.lock.RUnlock() + _, exists := r.accounts[address] + return exists +} + // 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. @@ -405,10 +444,6 @@ func verifyBlobRetrievals(t *testing.T, pool *BlobPool) { } } -func newReserver() *txpool.Reserver { - return txpool.NewReservationTracker().NewHandle(42) -} - // Tests that transactions can be loaded from disk on startup and that they are // correctly discarded if invalid. // diff --git a/core/txpool/legacypool/legacypool.go b/core/txpool/legacypool/legacypool.go index 75dc4a8461..04f1a2234c 100644 --- a/core/txpool/legacypool/legacypool.go +++ b/core/txpool/legacypool/legacypool.go @@ -237,7 +237,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 - reserver *txpool.Reserver // Address reserver to ensure exclusivity across subpools + 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 @@ -302,7 +302,7 @@ 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, reserver *txpool.Reserver) 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.reserver = reserver @@ -640,11 +640,18 @@ func (pool *LegacyPool) validateAuth(tx *types.Transaction) error { if err := pool.checkDelegationLimit(tx); err != nil { return err } - // Authorities must not conflict with any pending or queued transactions, - // nor with addresses that have already been reserved. + // For symmetry, allow at most one in-flight tx for any authority with a + // pending transaction. if auths := tx.SetCodeAuthorities(); len(auths) > 0 { for _, auth := range auths { - if pool.pending[auth] != nil || pool.queue[auth] != nil { + var count int + if pending := pool.pending[auth]; pending != nil { + count += pending.Len() + } + if queue := pool.queue[auth]; queue != nil { + count += queue.Len() + } + if count > 1 { return ErrAuthorityReserved } // Because there is no exclusive lock held between different subpools @@ -1907,9 +1914,14 @@ func (pool *LegacyPool) Clear() { // The transaction addition may attempt to reserve the sender addr which // 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) - pool.reserver.Release(senderAddr) + + for addr := range pool.pending { + if _, ok := pool.queue[addr]; !ok { + pool.reserver.Release(addr) + } + } + for addr := range pool.queue { + pool.reserver.Release(addr) } pool.all = newLookup() pool.priced = newPricedList(pool.all) diff --git a/core/txpool/legacypool/legacypool_test.go b/core/txpool/legacypool/legacypool_test.go index c47a655204..bb1323a7d1 100644 --- a/core/txpool/legacypool/legacypool_test.go +++ b/core/txpool/legacypool/legacypool_test.go @@ -24,6 +24,7 @@ import ( "math/big" "math/rand" "slices" + "sync" "sync/atomic" "testing" "time" @@ -171,8 +172,39 @@ func setupPool() (*LegacyPool, *ecdsa.PrivateKey) { return setupPoolWithConfig(params.TestChainConfig) } -func newReserver() *txpool.Reserver { - return txpool.NewReservationTracker().NewHandle(42) +// reserver is a utility struct to sanity check that accounts are +// properly reserved by the blobpool (no duplicate reserves or unreserves). +type reserver struct { + accounts map[common.Address]struct{} + lock sync.RWMutex +} + +func newReserver() txpool.Reserver { + return &reserver{accounts: make(map[common.Address]struct{})} +} + +func (r *reserver) Hold(addr common.Address) error { + r.lock.Lock() + defer r.lock.Unlock() + if _, exists := r.accounts[addr]; exists { + panic("already reserved") + } + r.accounts[addr] = struct{}{} + return nil +} + +func (r *reserver) Release(addr common.Address) error { + r.lock.Lock() + defer r.lock.Unlock() + if _, exists := r.accounts[addr]; !exists { + panic("not reserved") + } + delete(r.accounts, addr) + return nil +} + +func (r *reserver) Has(address common.Address) bool { + return false // reserver only supports a single pool } func setupPoolWithConfig(config *params.ChainConfig) (*LegacyPool, *ecdsa.PrivateKey) { @@ -2232,9 +2264,8 @@ func TestSetCodeTransactions(t *testing.T) { }{ { // Check that only one in-flight transaction is allowed for accounts - // with delegation set. Also verify the accepted transaction can be - // replaced by fee. - name: "only-one-in-flight", + // with delegation set. + name: "accept-one-inflight-tx-of-delegated-account", pending: 1, run: func(name string) { aa := common.Address{0xaa, 0xaa} @@ -2249,6 +2280,7 @@ 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) } + // Second and further transactions shall be rejected 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) } @@ -2260,6 +2292,70 @@ func TestSetCodeTransactions(t *testing.T) { if err := pool.addRemoteSync(pricedTransaction(0, 100000, big.NewInt(10), keyA)); err != nil { t.Fatalf("%s: failed to replace with remote transaction: %v", name, err) } + + // Reset the delegation, avoid leaking state into the other tests + statedb.SetCode(addrA, nil) + }, + }, + { + // This test is analogous to the previous one, but the delegation is pending + // instead of set. + name: "allow-one-tx-from-pooled-delegation", + pending: 2, + run: func(name string) { + // Create a pending delegation request from B. + if err := pool.addRemoteSync(setCodeTx(0, keyA, []unsignedAuth{{0, keyB}})); err != nil { + t.Fatalf("%s: failed to add with remote setcode transaction: %v", name, err) + } + // First transaction from B is accepted. + if err := pool.addRemoteSync(pricedTransaction(0, 100000, big.NewInt(1), keyB)); err != nil { + t.Fatalf("%s: failed to add remote transaction: %v", name, err) + } + // Second transaction fails due to limit. + if err := pool.addRemoteSync(pricedTransaction(1, 100000, big.NewInt(1), keyB)); !errors.Is(err, txpool.ErrInflightTxLimitReached) { + t.Fatalf("%s: error mismatch: want %v, have %v", name, txpool.ErrInflightTxLimitReached, err) + } + // Replace by fee for first transaction from B works. + if err := pool.addRemoteSync(pricedTransaction(0, 100000, big.NewInt(2), keyB)); err != nil { + t.Fatalf("%s: failed to add remote transaction: %v", name, err) + } + }, + }, + { + // This is the symmetric case of the previous one, where the delegation request + // is received after the transaction. The resulting state shall be the same. + name: "accept-authorization-from-sender-of-one-inflight-tx", + pending: 2, + run: func(name string) { + // The first in-flight transaction is accepted. + if err := pool.addRemoteSync(pricedTransaction(0, 100000, big.NewInt(1), keyB)); err != nil { + t.Fatalf("%s: failed to add with pending delegation: %v", name, err) + } + // Delegation is accepted. + if err := pool.addRemoteSync(setCodeTx(0, keyA, []unsignedAuth{{0, keyB}})); err != nil { + t.Fatalf("%s: failed to add remote transaction: %v", name, err) + } + // The second in-flight transaction is rejected. + if err := pool.addRemoteSync(pricedTransaction(1, 100000, big.NewInt(1), keyB)); !errors.Is(err, txpool.ErrInflightTxLimitReached) { + t.Fatalf("%s: error mismatch: want %v, have %v", name, txpool.ErrInflightTxLimitReached, err) + } + }, + }, + { + name: "reject-authorization-from-sender-with-more-than-one-inflight-tx", + pending: 2, + run: func(name string) { + // Submit two transactions. + if err := pool.addRemoteSync(pricedTransaction(0, 100000, big.NewInt(1), keyB)); err != nil { + t.Fatalf("%s: failed to add with pending delegation: %v", name, err) + } + if err := pool.addRemoteSync(pricedTransaction(1, 100000, big.NewInt(1), keyB)); err != nil { + t.Fatalf("%s: failed to add with pending delegation: %v", name, err) + } + // Delegation rejected since two txs are already in-flight. + if err := pool.addRemoteSync(setCodeTx(0, keyA, []unsignedAuth{{0, keyB}})); !errors.Is(err, ErrAuthorityReserved) { + t.Fatalf("%s: error mismatch: want %v, have %v", name, ErrAuthorityReserved, err) + } }, }, { @@ -2267,7 +2363,7 @@ func TestSetCodeTransactions(t *testing.T) { pending: 2, run: func(name string) { // Send two transactions where the first has no conflicting delegations and - // the second should be allowed despite conflicting with the authorities in 1). + // the second should be allowed despite conflicting with the authorities in the first. if err := pool.addRemoteSync(setCodeTx(0, keyA, []unsignedAuth{{1, keyC}})); err != nil { t.Fatalf("%s: failed to add with remote setcode transaction: %v", name, err) } @@ -2276,28 +2372,10 @@ func TestSetCodeTransactions(t *testing.T) { } }, }, - { - name: "allow-one-tx-from-pooled-delegation", - pending: 2, - run: func(name string) { - // Verify C cannot originate another transaction when it has a pooled delegation. - if err := pool.addRemoteSync(setCodeTx(0, keyA, []unsignedAuth{{0, keyC}})); err != nil { - t.Fatalf("%s: failed to add with remote setcode transaction: %v", name, err) - } - if err := pool.addRemoteSync(pricedTransaction(0, 100000, big.NewInt(1), keyC)); err != nil { - 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, txpool.ErrInflightTxLimitReached) { - t.Fatalf("%s: error mismatch: want %v, have %v", name, txpool.ErrInflightTxLimitReached, err) - } - }, - }, { name: "replace-by-fee-setcode-tx", pending: 1, run: func(name string) { - // 4. Fee bump the setcode tx send. if err := pool.addRemoteSync(setCodeTx(0, keyB, []unsignedAuth{{1, keyC}})); err != nil { t.Fatalf("%s: failed to add with remote setcode transaction: %v", name, err) } @@ -2307,44 +2385,85 @@ func TestSetCodeTransactions(t *testing.T) { }, }, { - name: "allow-tx-from-replaced-authority", - pending: 2, + name: "allow-more-than-one-tx-from-replaced-authority", + pending: 3, run: func(name string) { - // Fee bump with a different auth list. Make sure that unlocks the authorities. + // Send transaction from A with B as an authority. if err := pool.addRemoteSync(pricedSetCodeTx(0, 250000, uint256.NewInt(10), uint256.NewInt(3), keyA, []unsignedAuth{{0, keyB}})); err != nil { t.Fatalf("%s: failed to add with remote setcode transaction: %v", name, err) } + // Replace transaction with another having C as an authority. if err := pool.addRemoteSync(pricedSetCodeTx(0, 250000, uint256.NewInt(3000), uint256.NewInt(300), keyA, []unsignedAuth{{0, keyC}})); err != nil { t.Fatalf("%s: failed to add with remote setcode transaction: %v", name, err) } - // Now send a regular tx from B. + // B should not be considred as having an in-flight delegation, so + // should allow more than one pooled transaction. if err := pool.addRemoteSync(pricedTransaction(0, 100000, big.NewInt(10), keyB)); err != nil { t.Fatalf("%s: failed to replace with remote transaction: %v", name, err) } + if err := pool.addRemoteSync(pricedTransaction(1, 100000, big.NewInt(10), keyB)); err != nil { + t.Fatalf("%s: failed to replace with remote transaction: %v", name, err) + } }, }, { + // This test is analogous to the previous one, but the the replaced + // transaction is self-sponsored. name: "allow-tx-from-replaced-self-sponsor-authority", - pending: 2, + pending: 3, run: func(name string) { - // + // Send transaction from A with A as an authority. if err := pool.addRemoteSync(pricedSetCodeTx(0, 250000, uint256.NewInt(10), uint256.NewInt(3), keyA, []unsignedAuth{{0, keyA}})); err != nil { t.Fatalf("%s: failed to add with remote setcode transaction: %v", name, err) } + // Replace transaction with a transaction with B as an authority. if err := pool.addRemoteSync(pricedSetCodeTx(0, 250000, uint256.NewInt(30), uint256.NewInt(30), keyA, []unsignedAuth{{0, keyB}})); err != nil { t.Fatalf("%s: failed to add with remote setcode transaction: %v", name, err) } - // Now send a regular tx from keyA. - if err := pool.addRemoteSync(pricedTransaction(0, 100000, big.NewInt(1000), keyA)); err != nil { + // The one in-flight transaction limit from A no longer applies, so we + // can stack a second transaction for the account. + if err := pool.addRemoteSync(pricedTransaction(1, 100000, big.NewInt(1000), keyA)); err != nil { t.Fatalf("%s: failed to replace with remote transaction: %v", name, err) } - // Make sure we can still send from keyB. + // B should still be able to send transactions. if err := pool.addRemoteSync(pricedTransaction(0, 100000, big.NewInt(1000), keyB)); err != nil { t.Fatalf("%s: failed to replace with remote transaction: %v", name, err) } + // However B still has the limitation to one in-flight transaction. + if err := pool.addRemoteSync(pricedTransaction(1, 100000, big.NewInt(1), keyB)); !errors.Is(err, txpool.ErrInflightTxLimitReached) { + t.Fatalf("%s: error mismatch: want %v, have %v", name, txpool.ErrInflightTxLimitReached, err) + } }, }, { + name: "replacements-respect-inflight-tx-count", + pending: 2, + run: func(name string) { + // Send transaction from A with B as an authority. + if err := pool.addRemoteSync(pricedSetCodeTx(0, 250000, uint256.NewInt(10), uint256.NewInt(3), keyA, []unsignedAuth{{0, keyB}})); err != nil { + t.Fatalf("%s: failed to add with remote setcode transaction: %v", name, err) + } + // Send two transactions from B. Only the first should be accepted due + // to in-flight limit. + if err := pool.addRemoteSync(pricedTransaction(0, 100000, big.NewInt(1), keyB)); err != nil { + t.Fatalf("%s: failed to add remote transaction: %v", name, err) + } + if err := pool.addRemoteSync(pricedTransaction(1, 100000, big.NewInt(1), keyB)); !errors.Is(err, txpool.ErrInflightTxLimitReached) { + t.Fatalf("%s: error mismatch: want %v, have %v", name, txpool.ErrInflightTxLimitReached, err) + } + // Replace the in-flight transaction from B. + if err := pool.addRemoteSync(pricedTransaction(0, 100000, big.NewInt(30), keyB)); err != nil { + t.Fatalf("%s: failed to replace with remote transaction: %v", name, err) + } + // Ensure the in-flight limit for B is still in place. + if err := pool.addRemoteSync(pricedTransaction(1, 100000, big.NewInt(1), keyB)); !errors.Is(err, txpool.ErrInflightTxLimitReached) { + t.Fatalf("%s: error mismatch: want %v, have %v", name, txpool.ErrInflightTxLimitReached, err) + } + }, + }, + { + // Since multiple authorizations can be pending simultaneously, replacing + // one of them should not break the one in-flight-transaction limit. name: "track-multiple-conflicting-delegations", pending: 3, run: func(name string) { @@ -2369,19 +2488,6 @@ func TestSetCodeTransactions(t *testing.T) { } }, }, - { - name: "reject-delegation-from-pending-account", - pending: 1, - run: func(name string) { - // Attempt to submit a delegation from an account with a pending tx. - if err := pool.addRemoteSync(pricedTransaction(0, 100000, big.NewInt(1000), keyC)); err != nil { - t.Fatalf("%s: failed to add with remote setcode transaction: %v", name, err) - } - if err, want := pool.addRemoteSync(setCodeTx(0, keyA, []unsignedAuth{{1, keyC}})), ErrAuthorityReserved; !errors.Is(err, want) { - t.Fatalf("%s: error mismatch: want %v, have %v", name, want, err) - } - }, - }, { name: "remove-hash-from-authority-tracker", pending: 10, diff --git a/core/txpool/reserver.go b/core/txpool/reserver.go index 76ead0f3bb..b6ecef9f1a 100644 --- a/core/txpool/reserver.go +++ b/core/txpool/reserver.go @@ -52,22 +52,37 @@ func NewReservationTracker() *ReservationTracker { // NewHandle creates a named handle on the ReservationTracker. The handle // identifies the subpool so ownership of reservations can be determined. -func (r *ReservationTracker) NewHandle(id int) *Reserver { - return &Reserver{r, id} +func (r *ReservationTracker) NewHandle(id int) *ReservationHandle { + return &ReservationHandle{r, id} } -// Reserver is a named handle on ReservationTracker. It is held by subpools to +// Reserver is an interface for creating and releasing owned reservations in the +// ReservationTracker struct, which is shared between subpools. +type Reserver interface { + // Hold attempts to reserve the specified account address for the given pool. + // Returns an error if the account is already reserved. + Hold(addr common.Address) error + + // 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. + Release(addr common.Address) error + + // Has returns a flag indicating if the address has been reserved by a pool + // other than one with the current Reserver handle. + Has(address common.Address) bool +} + +// ReservationHandle is a named handle on ReservationTracker. It is held by subpools to // make reservations for accounts it is tracking. The id is used to determine // which pool owns an address and disallows non-owners to hold or release // addresses it doesn't own. -type Reserver struct { +type ReservationHandle struct { tracker *ReservationTracker id int } -// Hold attempts to reserve the specified account address for the given pool. -// Returns an error if the account is already reserved. -func (h *Reserver) Hold(addr common.Address) error { +// Hold implements the Reserver interface. +func (h *ReservationHandle) Hold(addr common.Address) error { h.tracker.lock.Lock() defer h.tracker.lock.Unlock() @@ -89,9 +104,8 @@ func (h *Reserver) Hold(addr common.Address) error { return nil } -// Release attempts to release the reservation for the specified account. -// Returns an error if the address is not reserved or is reserved by another pool. -func (h *Reserver) Release(addr common.Address) error { +// Release implements the Reserver interface. +func (h *ReservationHandle) Release(addr common.Address) error { h.tracker.lock.Lock() defer h.tracker.lock.Unlock() @@ -114,11 +128,11 @@ func (h *Reserver) Release(addr common.Address) error { return nil } -// Has returns a flag indicating if the address has been reserved or not. -func (h *Reserver) Has(address common.Address) bool { +// Has implements the Reserver interface. +func (h *ReservationHandle) Has(address common.Address) bool { h.tracker.lock.RLock() defer h.tracker.lock.RUnlock() - _, exists := h.tracker.accounts[address] - return exists + id, exists := h.tracker.accounts[address] + return exists && id != h.id } diff --git a/core/txpool/subpool.go b/core/txpool/subpool.go index 2cb5103875..8cfc14f164 100644 --- a/core/txpool/subpool.go +++ b/core/txpool/subpool.go @@ -105,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, reserver *Reserver) error + Init(gasTip uint64, head *types.Header, reserver Reserver) error // Close terminates any background processing threads and releases any held // resources. From 2547bb28a4a25969dd677f048155b64394fe59fd Mon Sep 17 00:00:00 2001 From: crStiv Date: Thu, 10 Apr 2025 12:26:35 +0300 Subject: [PATCH 08/12] eth/fetcher: Fix flaky TestTransactionForgotten test using mock clock (#31468) Fixes #31169 The TestTransactionForgotten test was flaky due to real time dependencies. This PR: - Replaces real time with mock clock for deterministic timing control - Adds precise state checks at timeout boundaries - Verifies underpriced cache states and cleanup - Improves test reliability by controlling transaction timestamps - Adds checks for transaction re-enqueueing behavior The changes ensure consistent test behavior without timing-related flakiness. --------- Co-authored-by: Zsolt Felfoldi --- eth/fetcher/tx_fetcher.go | 14 +-- eth/fetcher/tx_fetcher_test.go | 98 ++++++++++++++++++--- tests/fuzzers/txfetcher/txfetcher_fuzzer.go | 7 +- 3 files changed, 99 insertions(+), 20 deletions(-) diff --git a/eth/fetcher/tx_fetcher.go b/eth/fetcher/tx_fetcher.go index 97d1e29862..1c192d4112 100644 --- a/eth/fetcher/tx_fetcher.go +++ b/eth/fetcher/tx_fetcher.go @@ -202,22 +202,23 @@ type TxFetcher struct { fetchTxs func(string, []common.Hash) error // Retrieves a set of txs from a remote peer dropPeer func(string) // Drops a peer in case of announcement violation - step chan struct{} // Notification channel when the fetcher loop iterates - clock mclock.Clock // Time wrapper to simulate in tests - rand *mrand.Rand // Randomizer to use in tests instead of map range loops (soft-random) + step chan struct{} // Notification channel when the fetcher loop iterates + clock mclock.Clock // Monotonic clock or simulated clock for tests + realTime func() time.Time // Real system time or simulated time for tests + rand *mrand.Rand // Randomizer to use in tests instead of map range loops (soft-random) } // NewTxFetcher creates a transaction fetcher to retrieve transaction // based on hash announcements. func NewTxFetcher(hasTx func(common.Hash) bool, addTxs func([]*types.Transaction) []error, fetchTxs func(string, []common.Hash) error, dropPeer func(string)) *TxFetcher { - return NewTxFetcherForTests(hasTx, addTxs, fetchTxs, dropPeer, mclock.System{}, nil) + return NewTxFetcherForTests(hasTx, addTxs, fetchTxs, dropPeer, mclock.System{}, time.Now, nil) } // NewTxFetcherForTests is a testing method to mock out the realtime clock with // a simulated version and the internal randomness with a deterministic one. func NewTxFetcherForTests( hasTx func(common.Hash) bool, addTxs func([]*types.Transaction) []error, fetchTxs func(string, []common.Hash) error, dropPeer func(string), - clock mclock.Clock, rand *mrand.Rand) *TxFetcher { + clock mclock.Clock, realTime func() time.Time, rand *mrand.Rand) *TxFetcher { return &TxFetcher{ notify: make(chan *txAnnounce), cleanup: make(chan *txDelivery), @@ -237,6 +238,7 @@ func NewTxFetcherForTests( fetchTxs: fetchTxs, dropPeer: dropPeer, clock: clock, + realTime: realTime, rand: rand, } } @@ -293,7 +295,7 @@ func (f *TxFetcher) Notify(peer string, types []byte, sizes []uint32, hashes []c // isKnownUnderpriced reports whether a transaction hash was recently found to be underpriced. func (f *TxFetcher) isKnownUnderpriced(hash common.Hash) bool { prevTime, ok := f.underpriced.Peek(hash) - if ok && prevTime.Before(time.Now().Add(-maxTxUnderpricedTimeout)) { + if ok && prevTime.Before(f.realTime().Add(-maxTxUnderpricedTimeout)) { f.underpriced.Remove(hash) return false } diff --git a/eth/fetcher/tx_fetcher_test.go b/eth/fetcher/tx_fetcher_test.go index 52b3591086..7f3080f5f6 100644 --- a/eth/fetcher/tx_fetcher_test.go +++ b/eth/fetcher/tx_fetcher_test.go @@ -2150,9 +2150,22 @@ func containsHashInAnnounces(slice []announce, hash common.Hash) bool { return false } -// Tests that a transaction is forgotten after the timeout. +// TestTransactionForgotten verifies that underpriced transactions are properly +// forgotten after the timeout period, testing both the exact timeout boundary +// and the cleanup of the underpriced cache. func TestTransactionForgotten(t *testing.T) { - fetcher := NewTxFetcher( + // Test ensures that underpriced transactions are properly forgotten after a timeout period, + // including checks for timeout boundary and cache cleanup. + t.Parallel() + + // Create a mock clock for deterministic time control + mockClock := new(mclock.Simulated) + mockTime := func() time.Time { + nanoTime := int64(mockClock.Now()) + return time.Unix(nanoTime/1000000000, nanoTime%1000000000) + } + + fetcher := NewTxFetcherForTests( func(common.Hash) bool { return false }, func(txs []*types.Transaction) []error { errs := make([]error, len(txs)) @@ -2163,24 +2176,83 @@ func TestTransactionForgotten(t *testing.T) { }, func(string, []common.Hash) error { return nil }, func(string) {}, + mockClock, + mockTime, + rand.New(rand.NewSource(0)), // Use fixed seed for deterministic behavior ) fetcher.Start() defer fetcher.Stop() - // Create one TX which is 5 minutes old, and one which is recent - tx1 := types.NewTx(&types.LegacyTx{Nonce: 0}) - tx1.SetTime(time.Now().Add(-maxTxUnderpricedTimeout - 1*time.Second)) - tx2 := types.NewTx(&types.LegacyTx{Nonce: 1}) - // Enqueue both in the fetcher. They will be immediately tagged as underpriced - if err := fetcher.Enqueue("asdf", []*types.Transaction{tx1, tx2}, false); err != nil { + // Create two test transactions with the same timestamp + tx1 := types.NewTransaction(0, common.Address{}, big.NewInt(100), 21000, big.NewInt(1), nil) + tx2 := types.NewTransaction(1, common.Address{}, big.NewInt(100), 21000, big.NewInt(1), nil) + + now := mockTime() + tx1.SetTime(now) + tx2.SetTime(now) + + // Initial state: both transactions should be marked as underpriced + if err := fetcher.Enqueue("peer", []*types.Transaction{tx1, tx2}, false); err != nil { t.Fatal(err) } - // isKnownUnderpriced should trigger removal of the first tx (no longer be known underpriced) - if fetcher.isKnownUnderpriced(tx1.Hash()) { - t.Fatal("transaction should be forgotten by now") + if !fetcher.isKnownUnderpriced(tx1.Hash()) { + t.Error("tx1 should be underpriced") } - // isKnownUnderpriced should not trigger removal of the second if !fetcher.isKnownUnderpriced(tx2.Hash()) { - t.Fatal("transaction should be known underpriced") + t.Error("tx2 should be underpriced") + } + + // Verify cache size + if size := fetcher.underpriced.Len(); size != 2 { + t.Errorf("wrong underpriced cache size: got %d, want %d", size, 2) + } + + // Just before timeout: transactions should still be underpriced + mockClock.Run(maxTxUnderpricedTimeout - time.Second) + if !fetcher.isKnownUnderpriced(tx1.Hash()) { + t.Error("tx1 should still be underpriced before timeout") + } + if !fetcher.isKnownUnderpriced(tx2.Hash()) { + t.Error("tx2 should still be underpriced before timeout") + } + + // Exactly at timeout boundary: transactions should still be present + mockClock.Run(time.Second) + if !fetcher.isKnownUnderpriced(tx1.Hash()) { + t.Error("tx1 should be present exactly at timeout") + } + if !fetcher.isKnownUnderpriced(tx2.Hash()) { + t.Error("tx2 should be present exactly at timeout") + } + + // After timeout: transactions should be forgotten + mockClock.Run(time.Second) + if fetcher.isKnownUnderpriced(tx1.Hash()) { + t.Error("tx1 should be forgotten after timeout") + } + if fetcher.isKnownUnderpriced(tx2.Hash()) { + t.Error("tx2 should be forgotten after timeout") + } + + // Verify cache is empty + if size := fetcher.underpriced.Len(); size != 0 { + t.Errorf("wrong underpriced cache size after timeout: got %d, want 0", size) + } + + // Re-enqueue tx1 with updated timestamp + tx1.SetTime(mockTime()) + if err := fetcher.Enqueue("peer", []*types.Transaction{tx1}, false); err != nil { + t.Fatal(err) + } + if !fetcher.isKnownUnderpriced(tx1.Hash()) { + t.Error("tx1 should be underpriced after re-enqueueing with new timestamp") + } + if fetcher.isKnownUnderpriced(tx2.Hash()) { + t.Error("tx2 should remain forgotten") + } + + // Verify final cache state + if size := fetcher.underpriced.Len(); size != 1 { + t.Errorf("wrong final underpriced cache size: got %d, want 1", size) } } diff --git a/tests/fuzzers/txfetcher/txfetcher_fuzzer.go b/tests/fuzzers/txfetcher/txfetcher_fuzzer.go index 51f2fc3b4d..c136253a62 100644 --- a/tests/fuzzers/txfetcher/txfetcher_fuzzer.go +++ b/tests/fuzzers/txfetcher/txfetcher_fuzzer.go @@ -84,7 +84,12 @@ func fuzz(input []byte) int { }, func(string, []common.Hash) error { return nil }, nil, - clock, rand, + clock, + func() time.Time { + nanoTime := int64(clock.Now()) + return time.Unix(nanoTime/1000000000, nanoTime%1000000000) + }, + rand, ) f.Start() defer f.Stop() From 4906c9911316c52ccd20e122d1e092abe3c7d6ca Mon Sep 17 00:00:00 2001 From: HackyMiner Date: Thu, 10 Apr 2025 19:46:54 +0900 Subject: [PATCH 09/12] accounts/usbwallet: full 32bit chainId support for Trezor (#17439) This fix allows Trezor to support full 32bit chainId in geth, with the next version of firmware. For `chainId > 2147483630` case, Trezor returns signature bit only. - Trezor returns only signature parity for `chainId > 2147483630` case. - for `chainId == 2147483630` case, Trezor returns `MAX_UINT32` or `0`, but it doesn't matter. (`2147483630 * 2 + 35` = `4294967295`(`MAX_UINT32`)) chainId | returned signature_v | compatible issue ---------|------------------------|-------------------- 0 < chainId <= 255 | chainId * 2 + 35 + v | no issue (firmware `1.6.2` for Trezor one) 255 < chainId <= 2147483630 | chainId * 2 + 35 + v | ***fixed.*** *firmware `1.6.3`* chainId > 2147483630 | v | *firmware `1.6.3`* Please see also: full 32bit chainId support for Trezor - Trezor one: https://github.com/trezor/trezor-mcu/pull/399 ***merged*** - Trezor model T: https://github.com/trezor/trezor-core/pull/311 ***merged*** --------- Signed-off-by: Guillaume Ballet <3272758+gballet@users.noreply.github.com> Co-authored-by: Guillaume Ballet <3272758+gballet@users.noreply.github.com> --- accounts/usbwallet/trezor.go | 13 +++++++++++-- 1 file changed, 11 insertions(+), 2 deletions(-) diff --git a/accounts/usbwallet/trezor.go b/accounts/usbwallet/trezor.go index 1c4270d255..d4862d161b 100644 --- a/accounts/usbwallet/trezor.go +++ b/accounts/usbwallet/trezor.go @@ -25,6 +25,7 @@ import ( "errors" "fmt" "io" + "math" "math/big" "github.com/ethereum/go-ethereum/accounts" @@ -249,7 +250,11 @@ func (w *trezorDriver) trezorSign(derivationPath []uint32, tx *types.Transaction } } // Extract the Ethereum signature and do a sanity validation - if len(response.GetSignatureR()) == 0 || len(response.GetSignatureS()) == 0 || response.GetSignatureV() == 0 { + if len(response.GetSignatureR()) == 0 || len(response.GetSignatureS()) == 0 { + return common.Address{}, nil, errors.New("reply lacks signature") + } else if response.GetSignatureV() == 0 && int(chainID.Int64()) <= (math.MaxUint32-36)/2 { + // for chainId >= (MaxUint32-36)/2, Trezor returns signature bit only + // https://github.com/trezor/trezor-mcu/pull/399 return common.Address{}, nil, errors.New("reply lacks signature") } signature := append(append(response.GetSignatureR(), response.GetSignatureS()...), byte(response.GetSignatureV())) @@ -261,7 +266,11 @@ func (w *trezorDriver) trezorSign(derivationPath []uint32, tx *types.Transaction } else { // Trezor backend does not support typed transactions yet. signer = types.NewEIP155Signer(chainID) - signature[64] -= byte(chainID.Uint64()*2 + 35) + // if chainId is above (MaxUint32 - 36) / 2 then the final v values is returned + // directly. Otherwise, the returned value is 35 + chainid * 2. + if signature[64] > 1 && int(chainID.Int64()) <= (math.MaxUint32-36)/2 { + signature[64] -= byte(chainID.Uint64()*2 + 35) + } } // Inject the final signature into the transaction and sanity check the sender From 9b4eab6a29704f55fa7b4b92e296094f0dbcee22 Mon Sep 17 00:00:00 2001 From: jwasinger Date: Thu, 10 Apr 2025 19:49:54 +0800 Subject: [PATCH 10/12] eth/catalyst: in tests, manually sync txpool after initial chain insertion to prevent race between txpool head reset and promotion of txs that will be subsequently added (#31595) before this changes, this will result in numerous test failures: ``` > go test -run=Eth2AssembleBlock -c > stress ./catalyst.test ``` The reason is that after creating/inserting the test chain, there is a race between the txpool head reset and the promotion of txs added from tests. Ensuring that the txpool state is up to date with the head of the chain before proceeding fixes these flaky tests. --- eth/catalyst/api_test.go | 3 +++ 1 file changed, 3 insertions(+) diff --git a/eth/catalyst/api_test.go b/eth/catalyst/api_test.go index e91e07d05d..cb6ae053b6 100644 --- a/eth/catalyst/api_test.go +++ b/eth/catalyst/api_test.go @@ -447,6 +447,9 @@ func startEthService(t *testing.T, genesis *core.Genesis, blocks []*types.Block) n.Close() t.Fatal("can't import test blocks:", err) } + if err := ethservice.TxPool().Sync(); err != nil { + t.Fatal("failed to sync txpool after initial blockchain import:", err) + } ethservice.SetSynced() return n, ethservice From f64aa6eaf7a021226648aeae5ceef800fb9b66d0 Mon Sep 17 00:00:00 2001 From: jwasinger Date: Fri, 11 Apr 2025 00:21:32 +0800 Subject: [PATCH 11/12] internal/testlog: fix log output from sub-loggers (#31539) When we instantiate a sub-logger via `go-ethereum/internal/testlog/logger.With`, we copy the reference to the `bufHandler` from the parent logger. However, internally, `go-ethereum/internal/testlog/logger.With` calls `log/slog/Logger.With` which creates a new handler instance (via `internal/bufHandler.WithAttrs`). This PR modifies sub-logger instantiation to use the newly-instantiated handler, instead of copying the reference from the parent instance. The type cast from `slog.Handler` to `*bufHandler` in `internal/testlog/Logger.With` is safe here because a `internal/testlog/Logger` can only be instantiated with a `*bufHandler` as the underlying handler type. Note, that I've also removed a pre-existing method that broke the above assumption. However, this method is not used in our codebase. I'm not sure if the assumption holds for forks of geth (e.g. optimism has modified the testlogger somewhat allowing test loggers to accept arbitrary handler types), but it seems okay to break API compatibility given that this is in the `internal` package. closes https://github.com/ethereum/go-ethereum/issues/31533 --- internal/testlog/testlog.go | 28 +++++++------ internal/testlog/testlog_test.go | 67 ++++++++++++++++++++++++++++++++ tests/testdata | 2 +- 3 files changed, 81 insertions(+), 16 deletions(-) create mode 100644 internal/testlog/testlog_test.go diff --git a/internal/testlog/testlog.go b/internal/testlog/testlog.go index ad61af9eac..8a3ea85438 100644 --- a/internal/testlog/testlog.go +++ b/internal/testlog/testlog.go @@ -23,7 +23,6 @@ import ( "fmt" "log/slog" "sync" - "testing" "github.com/ethereum/go-ethereum/log" ) @@ -32,12 +31,21 @@ const ( termTimeFormat = "01-02|15:04:05.000" ) +// T wraps methods from testing.T used by the test logger into an interface. +// It is specified so that unit tests can instantiate the logger with an +// implementation of T which can capture the output of logging statements +// from T.Logf, as this cannot be using testing.T. +type T interface { + Logf(format string, args ...any) + Helper() +} + // logger implements log.Logger such that all output goes to the unit test log via // t.Logf(). All methods in between logger.Trace, logger.Debug, etc. are marked as test // helpers, so the file and line number in unit test output correspond to the call site // which emitted the log message. type logger struct { - t *testing.T + t T l log.Logger mu *sync.Mutex h *bufHandler @@ -78,7 +86,7 @@ func (h *bufHandler) WithGroup(_ string) slog.Handler { } // Logger returns a logger which logs to the unit test log of t. -func Logger(t *testing.T, level slog.Level) log.Logger { +func Logger(t T, level slog.Level) log.Logger { handler := bufHandler{ buf: []slog.Record{}, attrs: []slog.Attr{}, @@ -92,17 +100,6 @@ func Logger(t *testing.T, level slog.Level) log.Logger { } } -// LoggerWithHandler returns -func LoggerWithHandler(t *testing.T, handler slog.Handler) log.Logger { - var bh bufHandler - return &logger{ - t: t, - l: log.NewLogger(handler), - mu: new(sync.Mutex), - h: &bh, - } -} - func (l *logger) Handler() slog.Handler { return l.l.Handler() } @@ -170,7 +167,8 @@ func (l *logger) Crit(msg string, ctx ...interface{}) { } func (l *logger) With(ctx ...interface{}) log.Logger { - return &logger{l.t, l.l.With(ctx...), l.mu, l.h} + newLogger := l.l.With(ctx...) + return &logger{l.t, newLogger, l.mu, newLogger.Handler().(*bufHandler)} } func (l *logger) New(ctx ...interface{}) log.Logger { diff --git a/internal/testlog/testlog_test.go b/internal/testlog/testlog_test.go new file mode 100644 index 0000000000..7789a47270 --- /dev/null +++ b/internal/testlog/testlog_test.go @@ -0,0 +1,67 @@ +package testlog + +import ( + "bytes" + "fmt" + "io" + "strings" + "testing" + + "github.com/ethereum/go-ethereum/log" +) + +type mockT struct { + out io.Writer +} + +func (t *mockT) Helper() { + // noop for the purposes of unit tests +} + +func (t *mockT) Logf(format string, args ...any) { + // we could gate this operation in a mutex, but because testlogger + // only calls Logf with its internal mutex held, we just write output here + var lineBuf bytes.Buffer + if _, err := fmt.Fprintf(&lineBuf, format, args...); err != nil { + panic(err) + } + // The timestamp is locale-dependent, so we want to trim that off + // "INFO [01-01|00:00:00.000] a message ..." -> "a message..." + sanitized := strings.Split(lineBuf.String(), "]")[1] + if _, err := t.out.Write([]byte(sanitized)); err != nil { + panic(err) + } +} + +func TestLogging(t *testing.T) { + tests := []struct { + name string + expected string + run func(t *mockT) + }{ + { + "SubLogger", + ` Visible + Hide and seek foobar=123 + Also visible +`, + func(t *mockT) { + l := Logger(t, log.LevelInfo) + subLogger := l.New("foobar", 123) + + l.Info("Visible") + subLogger.Info("Hide and seek") + l.Info("Also visible") + }, + }, + } + + for _, tc := range tests { + outp := bytes.Buffer{} + mock := mockT{&outp} + tc.run(&mock) + if outp.String() != tc.expected { + fmt.Printf("output mismatch.\nwant: '%s'\ngot: '%s'\n", tc.expected, outp.String()) + } + } +} diff --git a/tests/testdata b/tests/testdata index 81862e4848..faf33b4714 160000 --- a/tests/testdata +++ b/tests/testdata @@ -1 +1 @@ -Subproject commit 81862e4848585a438d64f911a19b3825f0f4cd95 +Subproject commit faf33b471465d3c6cdc3d04fbd690895f78d33f2 From a9444ea312eb4077a470cdb9b394ec0544cf116b Mon Sep 17 00:00:00 2001 From: Felix Lange Date: Thu, 10 Apr 2025 23:54:44 +0200 Subject: [PATCH 12/12] tests/testdata: revert to v17.0 The submodule was accidentally updated to another commit by f64aa6eaf7. --- tests/testdata | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tests/testdata b/tests/testdata index faf33b4714..81862e4848 160000 --- a/tests/testdata +++ b/tests/testdata @@ -1 +1 @@ -Subproject commit faf33b471465d3c6cdc3d04fbd690895f78d33f2 +Subproject commit 81862e4848585a438d64f911a19b3825f0f4cd95