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

Merge branch 'ethereum:master' into gethintegration

Browse source
This commit is contained in:
Chen Kai 2025-04-14 22:18:13 +08:00 committed by GitHub
commit 4a5771aa2e
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
23 changed files with 791 additions and 204 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

11
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,8 +266,12 @@ func (w *trezorDriver) trezorSign(derivationPath []uint32, tx *types.Transaction
} else {
// Trezor backend does not support typed transactions yet.
signer = types.NewEIP155Signer(chainID)
// 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
signed, err := tx.WithSignature(signer, signature)

94
build/checksums.txt
View file

@ -5,54 +5,54 @@
# https://github.com/ethereum/execution-spec-tests/releases/download/pectra-devnet-6%40v1.0.0/
b69211752a3029083c020dc635fe12156ca1a6725a08559da540a0337586a77e fixtures_pectra-devnet-6.tar.gz
# version:golang 1.24.1
# version:golang 1.24.2
# https://go.dev/dl/
8244ebf46c65607db10222b5806aeb31c1fcf8979c1b6b12f60c677e9a3c0656 go1.24.1.src.tar.gz
8d627dc163a4bffa2b1887112ad6194af175dce108d606ed1714a089fb806033 go1.24.1.aix-ppc64.tar.gz
addbfce2056744962e2d7436313ab93486660cf7a2e066d171b9d6f2da7c7abe go1.24.1.darwin-amd64.tar.gz
58d529334561cff11087cd4ab18fe0b46d8d5aad88f45c02b9645f847e014512 go1.24.1.darwin-amd64.pkg
295581b5619acc92f5106e5bcb05c51869337eb19742fdfa6c8346c18e78ff88 go1.24.1.darwin-arm64.tar.gz
78b0fc8ddc344eb499f1a952c687cb84cbd28ba2b739cfa0d4eb042f07e44e82 go1.24.1.darwin-arm64.pkg
e70053f56f7eb93806d80cbd5726f78509a0a467602f7bea0e2c4ee8ed7c3968 go1.24.1.dragonfly-amd64.tar.gz
3595e2674ed8fe72e604ca59c964d3e5277aafb08475c2b1aaca2d2fd69c24fc go1.24.1.freebsd-386.tar.gz
47d7de8bb64d5c3ee7b6723aa62d5ecb11e3568ef2249bbe1d4bbd432d37c00c go1.24.1.freebsd-amd64.tar.gz
04eec3bcfaa14c1370cdf98e8307fac7e4853496c3045afb9c3124a29cbca205 go1.24.1.freebsd-arm.tar.gz
51aa70146e40cfdc20927424083dc86e6223f85dc08089913a1651973b55665b go1.24.1.freebsd-arm64.tar.gz
3c131d8e3fc285a1340f87813153e24226d3ddbd6e54f3facbd6e4c46a84655e go1.24.1.freebsd-riscv64.tar.gz
201d09da737ba39d5367f87d4e8b31edaeeb3dc9b9c407cb8cfb40f90c5a727a go1.24.1.illumos-amd64.tar.gz
8c530ecedbc17e42ce10177bea07ccc96a3e77c792ea1ea72173a9675d16ffa5 go1.24.1.linux-386.tar.gz
cb2396bae64183cdccf81a9a6df0aea3bce9511fc21469fb89a0c00470088073 go1.24.1.linux-amd64.tar.gz
8df5750ffc0281017fb6070fba450f5d22b600a02081dceef47966ffaf36a3af go1.24.1.linux-arm64.tar.gz
6d95f8d7884bfe2364644c837f080f2b585903d0b771eb5b06044e226a4f120a go1.24.1.linux-armv6l.tar.gz
19304a4a56e46d04604547d2d83235dc4f9b192c79832560ce337d26cc7b835a go1.24.1.linux-loong64.tar.gz
6347be77fa5359c12a5308c8ab87147c1fc4717b0c216493d1706c3b9fcde22d go1.24.1.linux-mips.tar.gz
1647df415f7030b82d4105670192aa7e8910e18563bb0d505192d72800cc2d21 go1.24.1.linux-mips64.tar.gz
762da594e4ec0f9cf6defae6ef971f5f7901203ee6a2d979e317adec96657317 go1.24.1.linux-mips64le.tar.gz
9d8133c7b23a557399fab870b5cf464079c2b623a43b214a7567cf11c254a444 go1.24.1.linux-mipsle.tar.gz
132f10999abbaccbada47fa85462db30c423955913b14d6c692de25f4636c766 go1.24.1.linux-ppc64.tar.gz
0fb522efcefabae6e37e69bdc444094e75bfe824ea6d4cc3cbc70c7ae1b16858 go1.24.1.linux-ppc64le.tar.gz
eaef4323d5467ff97fb1979c8491764060dade19f02f3275a9313f9a0da3b9c0 go1.24.1.linux-riscv64.tar.gz
6c05e14d8f11094cb56a1c50f390b6b658bed8a7cbd8d1a57e926581b7eabfce go1.24.1.linux-s390x.tar.gz
5dbb287d343ea00d58a70b11629f32ee716dc50a6875c459ea2018df0f294cd8 go1.24.1.netbsd-386.tar.gz
617aa3faee50ce84c343db0888e9a210c310a7203666b4ed620f31030c9fb32f go1.24.1.netbsd-amd64.tar.gz
59a928b7080c4a6ac985946274b7c65ce1cecc0b468ecd992d17b7c12fab9296 go1.24.1.netbsd-arm.tar.gz
28daa8d0feb4aef2af60cefa3305bb9314de7e8a05cbca41ac548964cdfe89b7 go1.24.1.netbsd-arm64.tar.gz
b7382b2f5d99813aeac14db482faa3bfbd47a68880b607fa2a7e669e26bab9cd go1.24.1.openbsd-386.tar.gz
2513b6537c45deead5e641c7ce7502913e7d5e6f0b21c52542fb11f81578690f go1.24.1.openbsd-amd64.tar.gz
853c1917d4fc7b144c55a02842aa48542d5cc798dde8db96dc0fdbc263200e04 go1.24.1.openbsd-arm.tar.gz
6bc207b91e6f6ae3347fb54616a8fb2f5c11983713846a4cef111ff3f4f94d14 go1.24.1.openbsd-arm64.tar.gz
4279260e2f2b94ee94e81470d13db7367f4393b061fee60985528fa0fa430df4 go1.24.1.openbsd-ppc64.tar.gz
6fc4023a0a339ee0778522364a127d94c78e62122288d47d820dba703f81dc07 go1.24.1.openbsd-riscv64.tar.gz
b5eb9fafd77146e7e1f748acfd95559580ecc8d2f15abf432a20f58c929c7cd2 go1.24.1.plan9-386.tar.gz
24dcad6361b141fc8cced15b092351e12a99d2e58d7013204a3013c50daf9fdd go1.24.1.plan9-amd64.tar.gz
a026ac3b55aa1e6fdc2aaab30207a117eafbe965ed81d3aa0676409f280ddc37 go1.24.1.plan9-arm.tar.gz
8e4f6a77388dc6e5aa481efd5abdb3b9f5c9463bb82f4db074494e04e5c84992 go1.24.1.solaris-amd64.tar.gz
b799f4ab264eef12a014c759383ed934056608c483e0f73e34ea6caf9f1df5f9 go1.24.1.windows-386.zip
db128981033ac82a64688a123f631e61297b6b8f52ca913145e57caa8ce94cc3 go1.24.1.windows-386.msi
95666b551453209a2b8869d29d177285ff9573af10f085d961d7ae5440f645ce go1.24.1.windows-amd64.zip
5968e7adcf26e68a54f1cd41ad561275a670a8e2ca5263bc375b524638557dfb go1.24.1.windows-amd64.msi
e28c4e6d0b913955765b46157ab88ae59bb636acaa12d7bec959aa6900f1cebd go1.24.1.windows-arm64.zip
6d352c1f154a102a5b90c480cc64bab205ccf2681e34e78a3a4d3f1ddfbc81e4 go1.24.1.windows-arm64.msi
9dc77ffadc16d837a1bf32d99c624cb4df0647cee7b119edd9e7b1bcc05f2e00 go1.24.2.src.tar.gz
427b373540d8fd51dbcc46bdecd340af109cd41514443c000d3dcde72b2c65a3 go1.24.2.aix-ppc64.tar.gz
238d9c065d09ff6af229d2e3b8b5e85e688318d69f4006fb85a96e41c216ea83 go1.24.2.darwin-amd64.tar.gz
535ed9ff283fee39575a7fb9b6d8b1901b6dc640d06dc71fd7d3faeefdaf8030 go1.24.2.darwin-amd64.pkg
b70f8b3c5b4ccb0ad4ffa5ee91cd38075df20fdbd953a1daedd47f50fbcff47a go1.24.2.darwin-arm64.tar.gz
4732f607a47ce4d898c0af01ff68f07e0820a6b50603aef5d5c777d1102505e2 go1.24.2.darwin-arm64.pkg
c17686b5fd61a663fbfafccfa177961be59386cf294e935ce35866b9dcb8e78a go1.24.2.dragonfly-amd64.tar.gz
026f1dd906189acff714c7625686bbc4ed91042618ba010d45b671461acc9e63 go1.24.2.freebsd-386.tar.gz
49399ba759b570a8f87d12179133403da6c2dd296d63a8830dee309161b9c40c go1.24.2.freebsd-amd64.tar.gz
1f48f47183794d97c29736004247ab541177cf984ac6322c78bc43828daa1172 go1.24.2.freebsd-arm.tar.gz
ef856428b60a8c0bd9a2cba596e83024be6f1c2d5574e89cb1ff2262b08df8b9 go1.24.2.freebsd-arm64.tar.gz
ec2088823e16df00600a6d0f72e9a7dc6d2f80c9c140c2043c0cf20e1404d1a9 go1.24.2.freebsd-riscv64.tar.gz
e030e7cedbb8688f1d75cb80f3de6ee2e6617a67d34051e794e5992b53462147 go1.24.2.illumos-amd64.tar.gz
4c382776d52313266f3026236297a224a6688751256a2dffa3f524d8d6f6c0ba go1.24.2.linux-386.tar.gz
68097bd680839cbc9d464a0edce4f7c333975e27a90246890e9f1078c7e702ad go1.24.2.linux-amd64.tar.gz
756274ea4b68fa5535eb9fe2559889287d725a8da63c6aae4d5f23778c229f4b go1.24.2.linux-arm64.tar.gz
438d5d3d7dcb239b58d893a715672eabe670b9730b1fd1c8fc858a46722a598a go1.24.2.linux-armv6l.tar.gz
6aefd3bf59c3c5592eda4fb287322207f119c2210f3795afa9be48d3ccb73e1b go1.24.2.linux-loong64.tar.gz
93e49bb4692783b0e9a2deab9558c6e8d2867f35592aeff285adda60924167f3 go1.24.2.linux-mips.tar.gz
6e86e703675016f3faf6604b8f68f20dc1bba75849136e6dd4f43f69c8a4a9d9 go1.24.2.linux-mips64.tar.gz
f233d237538ca1559a7d7cf519a29f0147923a951377bc4e467af4c059e68851 go1.24.2.linux-mips64le.tar.gz
545e1b9a7939f923fd53bde98334b987ef42eb353ee3e0bfede8aa06079d6b24 go1.24.2.linux-mipsle.tar.gz
6eab31481f2f46187bc1b6c887662eef06fc9d7271a8390854072cdb387c8d74 go1.24.2.linux-ppc64.tar.gz
5fff857791d541c71d8ea0171c73f6f99770d15ff7e2ad979104856d01f36563 go1.24.2.linux-ppc64le.tar.gz
91bda1558fcbd1c92769ad86c8f5cf796f8c67b0d9d9c19f76eecfc75ce71527 go1.24.2.linux-riscv64.tar.gz
1cb3448166d6abb515a85a3ee5afbdf932081fb58ad7143a8fb666fbc06146d9 go1.24.2.linux-s390x.tar.gz
a9a2c0db2e826f20f00b02bee01dfdaeb49591c2f6ffacb78dc64a950894f7ff go1.24.2.netbsd-386.tar.gz
cd1a35b76ed9c7b6c0c1616741bd319699a77867ade0be9924f32496c0a87a3f go1.24.2.netbsd-amd64.tar.gz
8c666388d066e479155cc5116950eeb435df28087ef277c18f1dc7479f836e60 go1.24.2.netbsd-arm.tar.gz
5d42f0be04f58da5be788a1e260f8747c316b8ce182bf0b273c2e4c691feaa1a go1.24.2.netbsd-arm64.tar.gz
688effa23ea3973cc8b0fdf4246712cbeef55ff20c45f3a9e28b0c2db04246cf go1.24.2.openbsd-386.tar.gz
e5daf95f1048d8026b1366450a3f8044d668b0639db6422ad9a83755c6745cf7 go1.24.2.openbsd-amd64.tar.gz
aeadaf74bd544d1a12ba9b14c0e7cdb1964de3ba9a52acb4619e91dbae7def7b go1.24.2.openbsd-arm.tar.gz
9e222d9adb0ce836a5b3c8d5aadbd167c8869c030b113f4a81aa88e9a200f279 go1.24.2.openbsd-arm64.tar.gz
192fffa34536adc3cd1bb7c1ee785b8bc156ae7afd10bbf5db99ec8f2e93066e go1.24.2.openbsd-ppc64.tar.gz
a23e90b451a390549042c2a7efbec6f29ed98b2d5618c8d2a35704e21be96e09 go1.24.2.openbsd-riscv64.tar.gz
5cdcafe455d859b02779611a5a1e1d63e498b922e05818fb3debe410a5959e9e go1.24.2.plan9-386.tar.gz
81351659804fa505c1b3ec6fdf9599f7f88df08614307eeb96071bf5e2e74beb go1.24.2.plan9-amd64.tar.gz
6e337d5def14ed0123423c1c32e2e6d8b19161e5d5ffaa7356dad48ee0fd80b4 go1.24.2.plan9-arm.tar.gz
07e6926ebc476c044d7d5b17706abfc52be52bccc2073d1734174efe63c6b35e go1.24.2.solaris-amd64.tar.gz
13d86cb818bba331da75fcd18246ab31a1067b44fb4a243b6dfd93097eda7f37 go1.24.2.windows-386.zip
8a702d9f7104a15bd935f4191c58c24c0b6389e066b9d5661b93915114a2bef0 go1.24.2.windows-386.msi
29c553aabee0743e2ffa3e9fa0cda00ef3b3cc4ff0bc92007f31f80fd69892e1 go1.24.2.windows-amd64.zip
acefb191e72fea0bdb1a3f5f8f6f5ab18b42b3bbce0c7183f189f25953aff275 go1.24.2.windows-amd64.msi
ab267f7f9a3366d48d7664be9e627ce3e63273231430cce5f7783fb910f14148 go1.24.2.windows-arm64.zip
d187bfe539356c39573d2f46766d1d08122b4f33da00fd14d12485fa9e241ff5 go1.24.2.windows-arm64.msi
# version:golangci 2.0.2
# https://github.com/golangci/golangci-lint/releases/

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

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

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

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

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

@ -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
@ -927,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 (
@ -1886,6 +1893,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()
@ -1904,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)

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

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

42
core/txpool/reserver.go
View file

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

2
core/txpool/subpool.go
View file

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

11
core/txpool/txpool.go
View file

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

7
eth/backend.go
View file

@ -76,6 +76,7 @@ type Ethereum struct {
handler *handler
discmix *enode.FairMix
dropper *dropper
// DB interfaces
chainDb ethdb.Database // Block chain database
@ -300,6 +301,8 @@ func New(stack *node.Node, config *ethconfig.Config) (*Ethereum, error) {
return nil, err
}
eth.dropper = newDropper(eth.p2pServer.MaxDialedConns(), eth.p2pServer.MaxInboundConns())
eth.miner = miner.New(eth, config.Miner, eth.engine)
eth.miner.SetExtra(makeExtraData(config.Miner.ExtraData))
eth.miner.SetPrioAddresses(config.TxPool.Locals)
@ -410,6 +413,9 @@ func (s *Ethereum) Start() error {
// Start the networking layer
s.handler.Start(s.p2pServer.MaxPeers)
// Start the connection manager
s.dropper.Start(s.p2pServer, func() bool { return !s.Synced() })
// start log indexer
s.filterMaps.Start()
go s.updateFilterMapsHeads()
@ -511,6 +517,7 @@ func (s *Ethereum) setupDiscovery() error {
func (s *Ethereum) Stop() error {
// Stop all the peer-related stuff first.
s.discmix.Close()
s.dropper.Stop()
s.handler.Stop()
// Then stop everything else.

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

167
eth/dropper.go Normal file
View file

@ -0,0 +1,167 @@
// 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 eth
import (
mrand "math/rand"
"slices"
"sync"
"time"
"github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/common/mclock"
"github.com/ethereum/go-ethereum/log"
"github.com/ethereum/go-ethereum/metrics"
"github.com/ethereum/go-ethereum/p2p"
)
const (
// Interval between peer drop events (uniform between min and max)
peerDropIntervalMin = 3 * time.Minute
// Interval between peer drop events (uniform between min and max)
peerDropIntervalMax = 7 * time.Minute
// Avoid dropping peers for some time after connection
doNotDropBefore = 10 * time.Minute
// How close to max should we initiate the drop timer. O should be fine,
// dropping when no more peers can be added. Larger numbers result in more
// aggressive drop behavior.
peerDropThreshold = 0
)
var (
// droppedInbound is the number of inbound peers dropped
droppedInbound = metrics.NewRegisteredMeter("eth/dropper/inbound", nil)
// droppedOutbound is the number of outbound peers dropped
droppedOutbound = metrics.NewRegisteredMeter("eth/dropper/outbound", nil)
)
// dropper monitors the state of the peer pool and makes changes as follows:
// - during sync the Downloader handles peer connections, so dropper is disabled
// - if not syncing and the peer count is close to the limit, it drops peers
// randomly every peerDropInterval to make space for new peers
// - peers are dropped separately from the inboud pool and from the dialed pool
type dropper struct {
maxDialPeers int // maximum number of dialed peers
maxInboundPeers int // maximum number of inbound peers
peersFunc getPeersFunc
syncingFunc getSyncingFunc
// peerDropTimer introduces churn if we are close to limit capacity.
// We handle Dialed and Inbound connections separately
peerDropTimer *time.Timer
wg sync.WaitGroup // wg for graceful shutdown
shutdownCh chan struct{}
}
// Callback type to get the list of connected peers.
type getPeersFunc func() []*p2p.Peer
// Callback type to get syncing status.
// Returns true while syncing, false when synced.
type getSyncingFunc func() bool
func newDropper(maxDialPeers, maxInboundPeers int) *dropper {
cm := &dropper{
maxDialPeers: maxDialPeers,
maxInboundPeers: maxInboundPeers,
peerDropTimer: time.NewTimer(randomDuration(peerDropIntervalMin, peerDropIntervalMax)),
shutdownCh: make(chan struct{}),
}
if peerDropIntervalMin > peerDropIntervalMax {
panic("peerDropIntervalMin duration must be less than or equal to peerDropIntervalMax duration")
}
return cm
}
// Start the dropper.
func (cm *dropper) Start(srv *p2p.Server, syncingFunc getSyncingFunc) {
cm.peersFunc = srv.Peers
cm.syncingFunc = syncingFunc
cm.wg.Add(1)
go cm.loop()
}
// Stop the dropper.
func (cm *dropper) Stop() {
cm.peerDropTimer.Stop()
close(cm.shutdownCh)
cm.wg.Wait()
}
// dropRandomPeer selects one of the peers randomly and drops it from the peer pool.
func (cm *dropper) dropRandomPeer() bool {
peers := cm.peersFunc()
var numInbound int
for _, p := range peers {
if p.Inbound() {
numInbound++
}
}
numDialed := len(peers) - numInbound
selectDoNotDrop := func(p *p2p.Peer) bool {
// Avoid dropping trusted and static peers, or recent peers.
// Only drop peers if their respective category (dialed/inbound)
// is close to limit capacity.
return p.Trusted() || p.StaticDialed() ||
p.Lifetime() < mclock.AbsTime(doNotDropBefore) ||
(p.DynDialed() && cm.maxDialPeers-numDialed > peerDropThreshold) ||
(p.Inbound() && cm.maxInboundPeers-numInbound > peerDropThreshold)
}
droppable := slices.DeleteFunc(peers, selectDoNotDrop)
if len(droppable) > 0 {
p := droppable[mrand.Intn(len(droppable))]
log.Debug("Dropping random peer", "inbound", p.Inbound(),
"id", p.ID(), "duration", common.PrettyDuration(p.Lifetime()), "peercountbefore", len(peers))
p.Disconnect(p2p.DiscUselessPeer)
if p.Inbound() {
droppedInbound.Mark(1)
} else {
droppedOutbound.Mark(1)
}
return true
}
return false
}
// randomDuration generates a random duration between min and max.
func randomDuration(min, max time.Duration) time.Duration {
if min > max {
panic("min duration must be less than or equal to max duration")
}
return time.Duration(mrand.Int63n(int64(max-min)) + int64(min))
}
// loop is the main loop of the connection dropper.
func (cm *dropper) loop() {
defer cm.wg.Done()
for {
select {
case <-cm.peerDropTimer.C:
// Drop a random peer if we are not syncing and the peer count is close to the limit.
if !cm.syncingFunc() {
cm.dropRandomPeer()
}
cm.peerDropTimer.Reset(randomDuration(peerDropIntervalMin, peerDropIntervalMax))
case <-cm.shutdownCh:
return
}
}
}

10
eth/fetcher/tx_fetcher.go
View file

@ -203,21 +203,22 @@ type TxFetcher struct {
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
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)
}
}

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

4
p2p/metrics.go
View file

@ -51,6 +51,10 @@ var (
dialSuccessMeter = metrics.NewRegisteredMeter("p2p/dials/success", nil)
dialConnectionError = metrics.NewRegisteredMeter("p2p/dials/error/connection", nil)
// count peers that stayed connected for at least 1 min
serve1MinSuccessMeter = metrics.NewRegisteredMeter("p2p/serves/success/1min", nil)
dial1MinSuccessMeter = metrics.NewRegisteredMeter("p2p/dials/success/1min", nil)
// handshake error meters
dialTooManyPeers = metrics.NewRegisteredMeter("p2p/dials/error/saturated", nil)
dialAlreadyConnected = metrics.NewRegisteredMeter("p2p/dials/error/known", nil)

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

34
p2p/peer.go
View file

@ -220,11 +220,35 @@ func (p *Peer) String() string {
return fmt.Sprintf("Peer %x %v", id[:8], p.RemoteAddr())
}
// Inbound returns true if the peer is an inbound connection
// Inbound returns true if the peer is an inbound (not dialed) connection.
func (p *Peer) Inbound() bool {
return p.rw.is(inboundConn)
}
// Trusted returns true if the peer is configured as trusted.
// Trusted peers are accepted in above the MaxInboundConns limit.
// The peer can be either inbound or dialed.
func (p *Peer) Trusted() bool {
return p.rw.is(trustedConn)
}
// DynDialed returns true if the peer was dialed successfully (passed handshake) and
// it is not configured as static.
func (p *Peer) DynDialed() bool {
return p.rw.is(dynDialedConn)
}
// StaticDialed returns true if the peer was dialed successfully (passed handshake) and
// it is configured as static.
func (p *Peer) StaticDialed() bool {
return p.rw.is(staticDialedConn)
}
// Lifetime returns the time since peer creation.
func (p *Peer) Lifetime() mclock.AbsTime {
return mclock.Now() - p.created
}
func newPeer(log log.Logger, conn *conn, protocols []Protocol) *Peer {
protomap := matchProtocols(protocols, conn.caps, conn)
p := &Peer{
@ -254,6 +278,8 @@ func (p *Peer) run() (remoteRequested bool, err error) {
p.wg.Add(2)
go p.readLoop(readErr)
go p.pingLoop()
live1min := time.NewTimer(1 * time.Minute)
defer live1min.Stop()
// Start all protocol handlers.
writeStart <- struct{}{}
@ -285,6 +311,12 @@ loop:
case err = <-p.disc:
reason = discReasonForError(err)
break loop
case <-live1min.C:
if p.Inbound() {
serve1MinSuccessMeter.Mark(1)
} else {
dial1MinSuccessMeter.Mark(1)
}
}
}

10
p2p/server.go
View file

@ -508,7 +508,7 @@ func (srv *Server) setupDiscovery() error {
func (srv *Server) setupDialScheduler() {
config := dialConfig{
self: srv.localnode.ID(),
maxDialPeers: srv.maxDialedConns(),
maxDialPeers: srv.MaxDialedConns(),
maxActiveDials: srv.MaxPendingPeers,
log: srv.Logger,
netRestrict: srv.NetRestrict,
@ -527,11 +527,11 @@ func (srv *Server) setupDialScheduler() {
}
}
func (srv *Server) maxInboundConns() int {
return srv.MaxPeers - srv.maxDialedConns()
func (srv *Server) MaxInboundConns() int {
return srv.MaxPeers - srv.MaxDialedConns()
}
func (srv *Server) maxDialedConns() (limit int) {
func (srv *Server) MaxDialedConns() (limit int) {
if srv.NoDial || srv.MaxPeers == 0 {
return 0
}
@ -736,7 +736,7 @@ func (srv *Server) postHandshakeChecks(peers map[enode.ID]*Peer, inboundCount in
switch {
case !c.is(trustedConn) && len(peers) >= srv.MaxPeers:
return DiscTooManyPeers
case !c.is(trustedConn) && c.is(inboundConn) && inboundCount >= srv.maxInboundConns():
case !c.is(trustedConn) && c.is(inboundConn) && inboundCount >= srv.MaxInboundConns():
return DiscTooManyPeers
case peers[c.node.ID()] != nil:
return DiscAlreadyConnected

27
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,15 +156,34 @@ 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 {
// 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.retries = 0
log = newLogger(m.protocol, int(p), m.port)
if int(p) != m.extPort {
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 {
@ -177,6 +198,8 @@ func (srv *Server) portMappingLoop() {
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()

2
version/version.go
View file

@ -19,6 +19,6 @@ package version
const (
Major = 1 // Major version component of the current release
Minor = 15 // Minor version component of the current release
Patch = 8 // Patch version component of the current release
Patch = 9 // Patch version component of the current release
Meta = "unstable" // Version metadata to append to the version string
)