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

Merge branch 'ethereum:master' into build/go1.24.2

Browse source
This commit is contained in:
levisyin 2025-04-11 12:15:26 +08:00 committed by GitHub
commit d2a9595061
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
24 changed files with 804 additions and 332 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

13
accounts/usbwallet/trezor.go
View file

@ -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

6
cmd/geth/chaincmd.go
View file

@ -79,11 +79,15 @@ if one is set. Otherwise it prints the genesis from the datadir.`,
Usage: "Import a blockchain file",
ArgsUsage: "<filename> (<filename 2> ... <filename N>) ",
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,

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

@ -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 (
@ -1311,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))
@ -1369,7 +1410,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 +1422,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 +1554,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]
@ -1754,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()
@ -1789,7 +1833,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)

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

@ -168,31 +168,42 @@ func (bc *testBlockChain) StateAt(common.Hash) (*state.StateDB, error) {
return bc.statedb, nil
}
// makeAddressReserver is a utility method to sanity check that accounts are
// reserver is a utility struct 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()
type reserver struct {
accounts map[common.Address]struct{}
lock sync.RWMutex
}
_, 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 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
@ -699,8 +710,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 +828,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 +929,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 +1008,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 +1109,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 +1552,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 +1649,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

4
core/txpool/errors.go
View file

@ -56,4 +56,8 @@ var (
// input transaction of non-blob type when a blob transaction from this sender
// remains pending (and vice-versa).
ErrAlreadyReserved = errors.New("address already reserved")
// ErrInflightTxLimitReached is returned when the maximum number of in-flight
// transactions is reached for specific accounts.
ErrInflightTxLimitReached = errors.New("in-flight transaction limit reached for delegated accounts")
)

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

@ -63,10 +63,6 @@ var (
// another remote transaction.
ErrTxPoolOverflow = errors.New("txpool is full")
// ErrInflightTxLimitReached is returned when the maximum number of in-flight
// transactions is reached for specific accounts.
ErrInflightTxLimitReached = errors.New("in-flight transaction limit reached for delegated accounts")
// ErrOutOfOrderTxFromDelegated is returned when the transaction with gapped
// nonce received from the accounts with delegation or pending delegation.
ErrOutOfOrderTxFromDelegated = errors.New("gapped-nonce tx from delegated accounts")
@ -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,10 +640,29 @@ 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.
// 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
// 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
}
}
@ -683,7 +698,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 +709,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)
}
}()
}
@ -919,8 +934,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 (
@ -1087,7 +1102,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 +1482,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 +1668,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 +1877,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.
@ -1876,12 +1893,15 @@ 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()
// 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 +1912,16 @@ 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)
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)
@ -1904,3 +1929,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)
}

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

@ -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)

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

@ -168,42 +168,52 @@ 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)
}
// 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) {
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 +346,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 +763,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 +973,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 +1025,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 +1186,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 +1285,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 +1318,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 +1362,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 +1467,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 +1616,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 +1706,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 +1909,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 +1976,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 +2167,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 +2240,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
@ -2254,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}
@ -2271,17 +2280,82 @@ 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)
// 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)
}
// 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 {
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)
}
},
},
{
@ -2289,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)
}
@ -2298,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, ErrInflightTxLimitReached) {
t.Fatalf("%s: error mismatch: want %v, have %v", name, 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)
}
@ -2329,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) {
@ -2386,20 +2483,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) {
t.Fatalf("%s: error mismatch: want %v, have %v", name, want, err)
}
},
},
{
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) {
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 +2538,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 +2573,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)

138
core/txpool/reserver.go Normal file
View file

@ -0,0 +1,138 @@
// Copyright 2025 The go-ethereum Authors
// This file is part of the go-ethereum library.
//
// The go-ethereum library is free software: you can redistribute it and/or modify
// it under the terms of the GNU Lesser General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// The go-ethereum library is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU Lesser General Public License for more details.
//
// You should have received a copy of the GNU Lesser General Public License
// along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
package txpool
import (
"errors"
"fmt"
"sync"
"github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/log"
"github.com/ethereum/go-ethereum/metrics"
)
var (
// reservationsGaugeName is the prefix of a per-subpool address reservation
// metric.
//
// This is mostly a sanity metric to ensure there's no bug that would make
// some subpool hog all the reservations due to mis-accounting.
reservationsGaugeName = "txpool/reservations"
)
// 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) *ReservationHandle {
return &ReservationHandle{r, id}
}
// 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 ReservationHandle struct {
tracker *ReservationTracker
id int
}
// Hold implements the Reserver interface.
func (h *ReservationHandle) 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 implements the Reserver interface.
func (h *ReservationHandle) 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 implements the Reserver interface.
func (h *ReservationHandle) Has(address common.Address) bool {
h.tracker.lock.RLock()
defer h.tracker.lock.RUnlock()
id, exists := h.tracker.accounts[address]
return exists && id != h.id
}

6
core/txpool/subpool.go
View file

@ -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.

88
core/txpool/txpool.go
View file

@ -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
@ -409,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
@ -562,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 {
@ -575,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

17
core/vm/instructions.go
View file

@ -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) {

2
core/vm/jump_table.go
View file

@ -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),

10
eth/backend.go
View file

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

3
eth/catalyst/api_test.go
View file

@ -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

14
eth/fetcher/tx_fetcher.go
View file

@ -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
}

98
eth/fetcher/tx_fetcher_test.go
View file

@ -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)
}
}

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

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

9
ethdb/remotedb/remotedb.go
View file

@ -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}
}

28
internal/testlog/testlog.go
View file

@ -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 {

67
internal/testlog/testlog_test.go Normal file
View file

@ -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())
}
}
}

49
p2p/nat/natupnp.go
View file

@ -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()

57
p2p/server_nat.go
View file

@ -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)
}
}
}

7
tests/fuzzers/txfetcher/txfetcher_fuzzer.go
View file

@ -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()