forks/go-ethereum
1
0
Fork 1
mirror of https://github.com/ethereum/go-ethereum.git synced 2026-08-19 18:32:23 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 1

Merge branch 'ethereum:master' into rlp-review

Browse source
This commit is contained in:
David Theodore 2024-05-23 20:10:11 -05:00 committed by GitHub
commit 642f6c11ab
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
43 changed files with 1759 additions and 927 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

4
.github/workflows/go.yml vendored
View file

@ -11,9 +11,9 @@ jobs:
build:
runs-on: self-hosted
steps:
- uses: actions/checkout@v2
- uses: actions/checkout@v4
- name: Set up Go
uses: actions/setup-go@v2
uses: actions/setup-go@v5
with:
go-version: 1.21.4
- name: Run tests

24
Makefile
View file

@ -2,31 +2,35 @@
# with Go source code. If you know what GOPATH is then you probably
# don't need to bother with make.
.PHONY: geth all test lint clean devtools help
.PHONY: geth all test lint fmt clean devtools help
GOBIN = ./build/bin
GO ?= latest
GORUN = go run
#? geth: Build geth
#? geth: Build geth.
geth:
$(GORUN) build/ci.go install ./cmd/geth
@echo "Done building."
@echo "Run \"$(GOBIN)/geth\" to launch geth."
#? all: Build all packages and executables
#? all: Build all packages and executables.
all:
$(GORUN) build/ci.go install
#? test: Run the tests
#? test: Run the tests.
test: all
$(GORUN) build/ci.go test
#? lint: Run certain pre-selected linters
#? lint: Run certain pre-selected linters.
lint: ## Run linters.
$(GORUN) build/ci.go lint
#? clean: Clean go cache, built executables, and the auto generated folder
#? fmt: Ensure consistent code formatting.
fmt:
gofmt -s -w $(shell find . -name "*.go")
#? clean: Clean go cache, built executables, and the auto generated folder.
clean:
go clean -cache
rm -fr build/_workspace/pkg/ $(GOBIN)/*
@ -34,7 +38,7 @@ clean:
# The devtools target installs tools required for 'go generate'.
# You need to put $GOBIN (or $GOPATH/bin) in your PATH to use 'go generate'.
#? devtools: Install recommended developer tools
#? devtools: Install recommended developer tools.
devtools:
env GOBIN= go install golang.org/x/tools/cmd/stringer@latest
env GOBIN= go install github.com/fjl/gencodec@latest
@ -45,5 +49,9 @@ devtools:
#? help: Get more info on make commands.
help: Makefile
@echo " Choose a command run in go-ethereum:"
@echo ''
@echo 'Usage:'
@echo ' make [target]'
@echo ''
@echo 'Targets:'
@sed -n 's/^#?//p' $< | column -t -s ':' | sort | sed -e 's/^/ /'

57
cmd/devp2p/discv4cmd.go
View file

@ -20,6 +20,7 @@ import (
"errors"
"fmt"
"net"
"net/http"
"strconv"
"strings"
"time"
@ -28,9 +29,11 @@ import (
"github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/crypto"
"github.com/ethereum/go-ethereum/internal/flags"
"github.com/ethereum/go-ethereum/log"
"github.com/ethereum/go-ethereum/p2p/discover"
"github.com/ethereum/go-ethereum/p2p/enode"
"github.com/ethereum/go-ethereum/params"
"github.com/ethereum/go-ethereum/rpc"
"github.com/urfave/cli/v2"
)
@ -45,6 +48,7 @@ var (
discv4ResolveJSONCommand,
discv4CrawlCommand,
discv4TestCommand,
discv4ListenCommand,
},
}
discv4PingCommand = &cli.Command{
@ -75,6 +79,14 @@ var (
Flags: discoveryNodeFlags,
ArgsUsage: "<nodes.json file>",
}
discv4ListenCommand = &cli.Command{
Name: "listen",
Usage: "Runs a discovery node",
Action: discv4Listen,
Flags: flags.Merge(discoveryNodeFlags, []cli.Flag{
httpAddrFlag,
}),
}
discv4CrawlCommand = &cli.Command{
Name: "crawl",
Usage: "Updates a nodes.json file with random nodes found in the DHT",
@ -131,6 +143,10 @@ var (
Usage: "Enode of the remote node under test",
EnvVars: []string{"REMOTE_ENODE"},
}
httpAddrFlag = &cli.StringFlag{
Name: "rpc",
Usage: "HTTP server listening address",
}
)
var discoveryNodeFlags = []cli.Flag{
@ -154,6 +170,27 @@ func discv4Ping(ctx *cli.Context) error {
return nil
}
func discv4Listen(ctx *cli.Context) error {
disc, _ := startV4(ctx)
defer disc.Close()
fmt.Println(disc.Self())
httpAddr := ctx.String(httpAddrFlag.Name)
if httpAddr == "" {
// Non-HTTP mode.
select {}
}
api := &discv4API{disc}
log.Info("Starting RPC API server", "addr", httpAddr)
srv := rpc.NewServer()
srv.RegisterName("discv4", api)
http.DefaultServeMux.Handle("/", srv)
httpsrv := http.Server{Addr: httpAddr, Handler: http.DefaultServeMux}
return httpsrv.ListenAndServe()
}
func discv4RequestRecord(ctx *cli.Context) error {
n := getNodeArg(ctx)
disc, _ := startV4(ctx)
@ -362,3 +399,23 @@ func parseBootnodes(ctx *cli.Context) ([]*enode.Node, error) {
}
return nodes, nil
}
type discv4API struct {
host *discover.UDPv4
}
func (api *discv4API) LookupRandom(n int) (ns []*enode.Node) {
it := api.host.RandomNodes()
for len(ns) < n && it.Next() {
ns = append(ns, it.Node())
}
return ns
}
func (api *discv4API) Buckets() [][]discover.BucketNode {
return api.host.TableBuckets()
}
func (api *discv4API) Self() *enode.Node {
return api.host.Self()
}

8
cmd/evm/t8n_test.go
View file

@ -234,7 +234,7 @@ func TestT8n(t *testing.T) {
{ // Test post-merge transition
base: "./testdata/24",
input: t8nInput{
"alloc.json", "txs.json", "env.json", "Merge", "",
"alloc.json", "txs.json", "env.json", "Paris", "",
},
output: t8nOutput{alloc: true, result: true},
expOut: "exp.json",
@ -242,7 +242,7 @@ func TestT8n(t *testing.T) {
{ // Test post-merge transition where input is missing random
base: "./testdata/24",
input: t8nInput{
"alloc.json", "txs.json", "env-missingrandom.json", "Merge", "",
"alloc.json", "txs.json", "env-missingrandom.json", "Paris", "",
},
output: t8nOutput{alloc: false, result: false},
expExitCode: 3,
@ -250,7 +250,7 @@ func TestT8n(t *testing.T) {
{ // Test base fee calculation
base: "./testdata/25",
input: t8nInput{
"alloc.json", "txs.json", "env.json", "Merge", "",
"alloc.json", "txs.json", "env.json", "Paris", "",
},
output: t8nOutput{alloc: true, result: true},
expOut: "exp.json",
@ -378,7 +378,7 @@ func TestT8nTracing(t *testing.T) {
{
base: "./testdata/32",
input: t8nInput{
"alloc.json", "txs.json", "env.json", "Merge", "",
"alloc.json", "txs.json", "env.json", "Paris", "",
},
extraArgs: []string{"--trace", "--trace.callframes"},
expectedTraces: []string{"trace-0-0x47806361c0fa084be3caa18afe8c48156747c01dbdfc1ee11b5aecdbe4fcf23e.jsonl"},

6
core/blockchain.go
View file

@ -1805,8 +1805,12 @@ func (bc *BlockChain) insertChain(chain types.Blocks, setHead bool) (int, error)
}
statedb.SetLogger(bc.logger)
// Enable prefetching to pull in trie node paths while processing transactions
// If we are past Byzantium, enable prefetching to pull in trie node paths
// while processing transactions. Before Byzantium the prefetcher is mostly
// useless due to the intermediate root hashing after each transaction.
if bc.chainConfig.IsByzantium(block.Number()) {
statedb.StartPrefetcher("chain")
}
activeState = statedb
// If we have a followup block, run that against the current state to pre-cache

3
core/chain_makers_test.go
View file

@ -43,12 +43,11 @@ func TestGeneratePOSChain(t *testing.T) {
bb = common.Address{0xbb}
funds = big.NewInt(0).Mul(big.NewInt(1337), big.NewInt(params.Ether))
config = *params.AllEthashProtocolChanges
asm4788 = common.Hex2Bytes("3373fffffffffffffffffffffffffffffffffffffffe14604d57602036146024575f5ffd5b5f35801560495762001fff810690815414603c575f5ffd5b62001fff01545f5260205ff35b5f5ffd5b62001fff42064281555f359062001fff015500")
gspec = &Genesis{
Config: &config,
Alloc: types.GenesisAlloc{
address: {Balance: funds},
params.BeaconRootsAddress: {Balance: common.Big0, Code: asm4788},
params.BeaconRootsAddress: {Code: params.BeaconRootsCode},
},
BaseFee: big.NewInt(params.InitialBaseFee),
Difficulty: common.Big1,

2
core/genesis.go
View file

@ -593,6 +593,8 @@ func DeveloperGenesisBlock(gasLimit uint64, faucet *common.Address) *Genesis {
common.BytesToAddress([]byte{7}): {Balance: big.NewInt(1)}, // ECScalarMul
common.BytesToAddress([]byte{8}): {Balance: big.NewInt(1)}, // ECPairing
common.BytesToAddress([]byte{9}): {Balance: big.NewInt(1)}, // BLAKE2b
// Pre-deploy EIP-4788 system contract
params.BeaconRootsAddress: types.Account{Nonce: 1, Code: params.BeaconRootsCode},
},
}
if faucet != nil {

92
core/state/state_object.go
View file

@ -21,6 +21,7 @@ import (
"fmt"
"io"
"maps"
"sync"
"time"
"github.com/ethereum/go-ethereum/common"
@ -33,6 +34,14 @@ import (
"github.com/holiman/uint256"
)
// hasherPool holds a pool of hashers used by state objects during concurrent
// trie updates.
var hasherPool = sync.Pool{
New: func() interface{} {
return crypto.NewKeccakState()
},
}
type Storage map[common.Hash]common.Hash
func (s Storage) Copy() Storage {
@ -118,16 +127,12 @@ func (s *stateObject) touch() {
}
}
// getTrie returns the associated storage trie. The trie will be opened
// if it's not loaded previously. An error will be returned if trie can't
// be loaded.
// getTrie returns the associated storage trie. The trie will be opened if it'
// not loaded previously. An error will be returned if trie can't be loaded.
//
// If a new trie is opened, it will be cached within the state object to allow
// subsequent reads to expand the same trie instead of reloading from disk.
func (s *stateObject) getTrie() (Trie, error) {
if s.trie == nil {
// Try fetching from prefetcher first
if s.data.Root != types.EmptyRootHash && s.db.prefetcher != nil {
// When the miner is creating the pending state, there is no prefetcher
s.trie = s.db.prefetcher.trie(s.addrHash, s.data.Root)
}
if s.trie == nil {
tr, err := s.db.db.OpenStorageTrie(s.db.originalRoot, s.address, s.data.Root, s.db.trie)
if err != nil {
@ -135,10 +140,26 @@ func (s *stateObject) getTrie() (Trie, error) {
}
s.trie = tr
}
}
return s.trie, nil
}
// getPrefetchedTrie returns the associated trie, as populated by the prefetcher
// if it's available.
//
// Note, opposed to getTrie, this method will *NOT* blindly cache the resulting
// trie in the state object. The caller might want to do that, but it's cleaner
// to break the hidden interdependency between retrieving tries from the db or
// from the prefetcher.
func (s *stateObject) getPrefetchedTrie() (Trie, error) {
// If there's nothing to meaningfully return, let the user figure it out by
// pulling the trie from disk.
if s.data.Root == types.EmptyRootHash || s.db.prefetcher == nil {
return nil, nil
}
// Attempt to retrieve the trie from the pretecher
return s.db.prefetcher.trie(s.addrHash, s.data.Root)
}
// GetState retrieves a value from the account storage trie.
func (s *stateObject) GetState(key common.Hash) common.Hash {
value, _ := s.getState(key)
@ -248,7 +269,7 @@ func (s *stateObject) setState(key common.Hash, value common.Hash, origin common
// finalise moves all dirty storage slots into the pending area to be hashed or
// committed later. It is invoked at the end of every transaction.
func (s *stateObject) finalise(prefetch bool) {
func (s *stateObject) finalise() {
slotsToPrefetch := make([][]byte, 0, len(s.dirtyStorage))
for key, value := range s.dirtyStorage {
// If the slot is different from its original value, move it into the
@ -263,8 +284,10 @@ func (s *stateObject) finalise(prefetch bool) {
delete(s.pendingStorage, key)
}
}
if s.db.prefetcher != nil && prefetch && len(slotsToPrefetch) > 0 && s.data.Root != types.EmptyRootHash {
s.db.prefetcher.prefetch(s.addrHash, s.data.Root, s.address, slotsToPrefetch)
if s.db.prefetcher != nil && len(slotsToPrefetch) > 0 && s.data.Root != types.EmptyRootHash {
if err := s.db.prefetcher.prefetch(s.addrHash, s.data.Root, s.address, slotsToPrefetch); err != nil {
log.Error("Failed to prefetch slots", "addr", s.address, "slots", len(slotsToPrefetch), "err", err)
}
}
if len(s.dirtyStorage) > 0 {
s.dirtyStorage = make(Storage)
@ -281,27 +304,44 @@ func (s *stateObject) finalise(prefetch bool) {
// loading or updating of the trie, an error will be returned. Furthermore,
// this function will return the mutated storage trie, or nil if there is no
// storage change at all.
//
// It assumes all the dirty storage slots have been finalized before.
func (s *stateObject) updateTrie() (Trie, error) {
// Make sure all dirty slots are finalized into the pending storage area
s.finalise(false)
// Short circuit if nothing changed, don't bother with hashing anything
if len(s.pendingStorage) == 0 {
return s.trie, nil
}
// Retrieve a pretecher populated trie, or fall back to the database
tr, err := s.getPrefetchedTrie()
switch {
case err != nil:
// Fetcher retrieval failed, something's very wrong, abort
s.db.setError(err)
return nil, err
case tr == nil:
// Fetcher not running or empty trie, fallback to the database trie
tr, err = s.getTrie()
if err != nil {
s.db.setError(err)
return nil, err
}
default:
// Prefetcher returned a live trie, swap it out for the current one
s.trie = tr
}
// The snapshot storage map for the object
var (
storage map[common.Hash][]byte
origin map[common.Hash][]byte
)
tr, err := s.getTrie()
if err != nil {
s.db.setError(err)
return nil, err
}
// Insert all the pending storage updates into the trie
usedStorage := make([][]byte, 0, len(s.pendingStorage))
hasher := hasherPool.Get().(crypto.KeccakState)
defer hasherPool.Put(hasher)
// Perform trie updates before deletions. This prevents resolution of unnecessary trie nodes
// in circumstances similar to the following:
//
@ -330,26 +370,30 @@ func (s *stateObject) updateTrie() (Trie, error) {
s.db.setError(err)
return nil, err
}
s.db.StorageUpdated += 1
s.db.StorageUpdated.Add(1)
} else {
deletions = append(deletions, key)
}
// Cache the mutated storage slots until commit
if storage == nil {
s.db.storagesLock.Lock()
if storage = s.db.storages[s.addrHash]; storage == nil {
storage = make(map[common.Hash][]byte)
s.db.storages[s.addrHash] = storage
}
s.db.storagesLock.Unlock()
}
khash := crypto.HashData(s.db.hasher, key[:])
khash := crypto.HashData(hasher, key[:])
storage[khash] = encoded // encoded will be nil if it's deleted
// Cache the original value of mutated storage slots
if origin == nil {
s.db.storagesLock.Lock()
if origin = s.db.storagesOrigin[s.address]; origin == nil {
origin = make(map[common.Hash][]byte)
s.db.storagesOrigin[s.address] = origin
}
s.db.storagesLock.Unlock()
}
// Track the original value of slot only if it's mutated first time
if _, ok := origin[khash]; !ok {
@ -369,7 +413,7 @@ func (s *stateObject) updateTrie() (Trie, error) {
s.db.setError(err)
return nil, err
}
s.db.StorageDeleted += 1
s.db.StorageDeleted.Add(1)
}
// If no slots were touched, issue a warning as we shouldn't have done all
// the above work in the first place

104
core/state/statedb.go
View file

@ -24,6 +24,7 @@ import (
"slices"
"sort"
"sync"
"sync/atomic"
"time"
"github.com/ethereum/go-ethereum/common"
@ -97,9 +98,11 @@ type StateDB struct {
// These maps hold the state changes (including the corresponding
// original value) that occurred in this **block**.
accounts map[common.Hash][]byte // The mutated accounts in 'slim RLP' encoding
storages map[common.Hash]map[common.Hash][]byte // The mutated slots in prefix-zero trimmed rlp format
accountsOrigin map[common.Address][]byte // The original value of mutated accounts in 'slim RLP' encoding
storages map[common.Hash]map[common.Hash][]byte // The mutated slots in prefix-zero trimmed rlp format
storagesOrigin map[common.Address]map[common.Hash][]byte // The original value of mutated slots in prefix-zero trimmed rlp format
storagesLock sync.Mutex // Mutex protecting the maps during concurrent updates/commits
// This map holds 'live' objects, which will get modified while
// processing a state transition.
@ -165,9 +168,9 @@ type StateDB struct {
TrieDBCommits time.Duration
AccountUpdated int
StorageUpdated int
StorageUpdated atomic.Int64
AccountDeleted int
StorageDeleted int
StorageDeleted atomic.Int64
// Testing hooks
onCommit func(states *triestate.Set) // Hook invoked when commit is performed
@ -214,7 +217,8 @@ func (s *StateDB) SetLogger(l *tracing.Hooks) {
// commit phase, most of the needed data is already hot.
func (s *StateDB) StartPrefetcher(namespace string) {
if s.prefetcher != nil {
s.prefetcher.close()
s.prefetcher.terminate(false)
s.prefetcher.report()
s.prefetcher = nil
}
if s.snap != nil {
@ -226,7 +230,8 @@ func (s *StateDB) StartPrefetcher(namespace string) {
// from the gathered metrics.
func (s *StateDB) StopPrefetcher() {
if s.prefetcher != nil {
s.prefetcher.close()
s.prefetcher.terminate(false)
s.prefetcher.report()
s.prefetcher = nil
}
}
@ -544,9 +549,6 @@ func (s *StateDB) GetTransientState(addr common.Address, key common.Hash) common
// updateStateObject writes the given object to the trie.
func (s *StateDB) updateStateObject(obj *stateObject) {
// Track the amount of time wasted on updating the account from the trie
defer func(start time.Time) { s.AccountUpdates += time.Since(start) }(time.Now())
// Encode the account and update the account trie
addr := obj.Address()
if err := s.trie.UpdateAccount(addr, &obj.data); err != nil {
@ -575,10 +577,6 @@ func (s *StateDB) updateStateObject(obj *stateObject) {
// deleteStateObject removes the given object from the state trie.
func (s *StateDB) deleteStateObject(addr common.Address) {
// Track the amount of time wasted on deleting the account from the trie
defer func(start time.Time) { s.AccountUpdates += time.Since(start) }(time.Now())
// Delete the account from the trie
if err := s.trie.DeleteAccount(addr); err != nil {
s.setError(fmt.Errorf("deleteStateObject (%x) error: %v", addr[:], err))
}
@ -743,13 +741,6 @@ func (s *StateDB) Copy() *StateDB {
// in the middle of a transaction.
state.accessList = s.accessList.Copy()
state.transientStorage = s.transientStorage.Copy()
// If there's a prefetcher running, make an inactive copy of it that can
// only access data but does not actively preload (since the user will not
// know that they need to explicitly terminate an active copy).
if s.prefetcher != nil {
state.prefetcher = s.prefetcher.copy()
}
return state
}
@ -820,7 +811,7 @@ func (s *StateDB) Finalise(deleteEmptyObjects bool) {
delete(s.accountsOrigin, obj.address) // Clear out any previously updated account data (may be recreated via a resurrect)
delete(s.storagesOrigin, obj.address) // Clear out any previously updated storage data (may be recreated via a resurrect)
} else {
obj.finalise(true) // Prefetch slots in the background
obj.finalise()
s.markUpdate(addr)
}
// At this point, also ship the address off to the precacher. The precacher
@ -829,7 +820,9 @@ func (s *StateDB) Finalise(deleteEmptyObjects bool) {
addressesToPrefetch = append(addressesToPrefetch, common.CopyBytes(addr[:])) // Copy needed for closure
}
if s.prefetcher != nil && len(addressesToPrefetch) > 0 {
s.prefetcher.prefetch(common.Hash{}, s.originalRoot, common.Address{}, addressesToPrefetch)
if err := s.prefetcher.prefetch(common.Hash{}, s.originalRoot, common.Address{}, addressesToPrefetch); err != nil {
log.Error("Failed to prefetch addresses", "addresses", len(addressesToPrefetch), "err", err)
}
}
// Invalidate journal because reverting across transactions is not allowed.
s.clearJournalAndRefund()
@ -842,42 +835,52 @@ func (s *StateDB) IntermediateRoot(deleteEmptyObjects bool) common.Hash {
// Finalise all the dirty storage states and write them into the tries
s.Finalise(deleteEmptyObjects)
// If there was a trie prefetcher operating, it gets aborted and irrevocably
// modified after we start retrieving tries. Remove it from the statedb after
// this round of use.
//
// This is weird pre-byzantium since the first tx runs with a prefetcher and
// the remainder without, but pre-byzantium even the initial prefetcher is
// useless, so no sleep lost.
prefetcher := s.prefetcher
// If there was a trie prefetcher operating, terminate it async so that the
// individual storage tries can be updated as soon as the disk load finishes.
if s.prefetcher != nil {
s.prefetcher.terminate(true)
defer func() {
s.prefetcher.close()
s.prefetcher = nil
s.prefetcher.report()
s.prefetcher = nil // Pre-byzantium, unset any used up prefetcher
}()
}
// Although naively it makes sense to retrieve the account trie and then do
// the contract storage and account updates sequentially, that short circuits
// the account prefetcher. Instead, let's process all the storage updates
// first, giving the account prefetches just a few more milliseconds of time
// to pull useful data from disk.
start := time.Now()
// Process all storage updates concurrently. The state object update root
// method will internally call a blocking trie fetch from the prefetcher,
// so there's no need to explicitly wait for the prefetchers to finish.
var (
start = time.Now()
workers errgroup.Group
)
if s.db.TrieDB().IsVerkle() {
// Whilst MPT storage tries are independent, Verkle has one single trie
// for all the accounts and all the storage slots merged together. The
// former can thus be simply parallelized, but updating the latter will
// need concurrency support within the trie itself. That's a TODO for a
// later time.
workers.SetLimit(1)
}
for addr, op := range s.mutations {
if op.applied {
if op.applied || op.isDelete() {
continue
}
if op.isDelete() {
continue
}
s.stateObjects[addr].updateRoot()
obj := s.stateObjects[addr] // closure for the task runner below
workers.Go(func() error {
obj.updateRoot()
return nil
})
}
workers.Wait()
s.StorageUpdates += time.Since(start)
// Now we're about to start to write changes to the trie. The trie is so far
// _untouched_. We can check with the prefetcher, if it can give us a trie
// which has the same root, but also has some content loaded into it.
if prefetcher != nil {
if trie := prefetcher.trie(common.Hash{}, s.originalRoot); trie != nil {
start = time.Now()
if s.prefetcher != nil {
if trie, err := s.prefetcher.trie(common.Hash{}, s.originalRoot); err != nil {
log.Error("Failed to retrieve account pre-fetcher trie", "err", err)
} else if trie != nil {
s.trie = trie
}
}
@ -913,8 +916,10 @@ func (s *StateDB) IntermediateRoot(deleteEmptyObjects bool) common.Hash {
s.deleteStateObject(deletedAddr)
s.AccountDeleted += 1
}
if prefetcher != nil {
prefetcher.used(common.Hash{}, s.originalRoot, usedAddrs)
s.AccountUpdates += time.Since(start)
if s.prefetcher != nil {
s.prefetcher.used(common.Hash{}, s.originalRoot, usedAddrs)
}
// Track the amount of time wasted on hashing the account trie
defer func(start time.Time) { s.AccountHashes += time.Since(start) }(time.Now())
@ -1255,15 +1260,16 @@ func (s *StateDB) Commit(block uint64, deleteEmptyObjects bool) (common.Hash, er
return common.Hash{}, err
}
accountUpdatedMeter.Mark(int64(s.AccountUpdated))
storageUpdatedMeter.Mark(int64(s.StorageUpdated))
storageUpdatedMeter.Mark(s.StorageUpdated.Load())
accountDeletedMeter.Mark(int64(s.AccountDeleted))
storageDeletedMeter.Mark(int64(s.StorageDeleted))
storageDeletedMeter.Mark(s.StorageDeleted.Load())
accountTrieUpdatedMeter.Mark(int64(accountTrieNodesUpdated))
accountTrieDeletedMeter.Mark(int64(accountTrieNodesDeleted))
storageTriesUpdatedMeter.Mark(int64(storageTrieNodesUpdated))
storageTriesDeletedMeter.Mark(int64(storageTrieNodesDeleted))
s.AccountUpdated, s.AccountDeleted = 0, 0
s.StorageUpdated, s.StorageDeleted = 0, 0
s.StorageUpdated.Store(0)
s.StorageDeleted.Store(0)
// If snapshotting is enabled, update the snapshot tree with this new version
if s.snap != nil {

254
core/state/trie_prefetcher.go
View file

@ -17,6 +17,7 @@
package state
import (
"errors"
"sync"
"github.com/ethereum/go-ethereum/common"
@ -27,6 +28,10 @@ import (
var (
// triePrefetchMetricsPrefix is the prefix under which to publish the metrics.
triePrefetchMetricsPrefix = "trie/prefetch/"
// errTerminated is returned if a fetcher is attempted to be operated after it
// has already terminated.
errTerminated = errors.New("fetcher is already terminated")
)
// triePrefetcher is an active prefetcher, which receives accounts or storage
@ -37,52 +42,64 @@ var (
type triePrefetcher struct {
db Database // Database to fetch trie nodes through
root common.Hash // Root hash of the account trie for metrics
fetches map[string]Trie // Partially or fully fetched tries. Only populated for inactive copies.
fetchers map[string]*subfetcher // Subfetchers for each trie
term chan struct{} // Channel to signal interruption
deliveryMissMeter metrics.Meter
accountLoadMeter metrics.Meter
accountDupMeter metrics.Meter
accountSkipMeter metrics.Meter
accountWasteMeter metrics.Meter
storageLoadMeter metrics.Meter
storageDupMeter metrics.Meter
storageSkipMeter metrics.Meter
storageWasteMeter metrics.Meter
}
func newTriePrefetcher(db Database, root common.Hash, namespace string) *triePrefetcher {
prefix := triePrefetchMetricsPrefix + namespace
p := &triePrefetcher{
return &triePrefetcher{
db: db,
root: root,
fetchers: make(map[string]*subfetcher), // Active prefetchers use the fetchers map
term: make(chan struct{}),
deliveryMissMeter: metrics.GetOrRegisterMeter(prefix+"/deliverymiss", nil),
accountLoadMeter: metrics.GetOrRegisterMeter(prefix+"/account/load", nil),
accountDupMeter: metrics.GetOrRegisterMeter(prefix+"/account/dup", nil),
accountSkipMeter: metrics.GetOrRegisterMeter(prefix+"/account/skip", nil),
accountWasteMeter: metrics.GetOrRegisterMeter(prefix+"/account/waste", nil),
storageLoadMeter: metrics.GetOrRegisterMeter(prefix+"/storage/load", nil),
storageDupMeter: metrics.GetOrRegisterMeter(prefix+"/storage/dup", nil),
storageSkipMeter: metrics.GetOrRegisterMeter(prefix+"/storage/skip", nil),
storageWasteMeter: metrics.GetOrRegisterMeter(prefix+"/storage/waste", nil),
}
return p
}
// close iterates over all the subfetchers, aborts any that were left spinning
// and reports the stats to the metrics subsystem.
func (p *triePrefetcher) close() {
// terminate iterates over all the subfetchers and issues a terminateion request
// to all of them. Depending on the async parameter, the method will either block
// until all subfetchers spin down, or return immediately.
func (p *triePrefetcher) terminate(async bool) {
// Short circuit if the fetcher is already closed
select {
case <-p.term:
return
default:
}
// Termiante all sub-fetchers, sync or async, depending on the request
for _, fetcher := range p.fetchers {
fetcher.abort() // safe to do multiple times
fetcher.terminate(async)
}
close(p.term)
}
// report aggregates the pre-fetching and usage metrics and reports them.
func (p *triePrefetcher) report() {
if !metrics.Enabled {
return
}
for _, fetcher := range p.fetchers {
fetcher.wait() // ensure the fetcher's idle before poking in its internals
if metrics.Enabled {
if fetcher.root == p.root {
p.accountLoadMeter.Mark(int64(len(fetcher.seen)))
p.accountDupMeter.Mark(int64(fetcher.dups))
p.accountSkipMeter.Mark(int64(len(fetcher.tasks)))
for _, key := range fetcher.used {
delete(fetcher.seen, string(key))
}
@ -90,107 +107,61 @@ func (p *triePrefetcher) close() {
} else {
p.storageLoadMeter.Mark(int64(len(fetcher.seen)))
p.storageDupMeter.Mark(int64(fetcher.dups))
p.storageSkipMeter.Mark(int64(len(fetcher.tasks)))
for _, key := range fetcher.used {
delete(fetcher.seen, string(key))
}
p.storageWasteMeter.Mark(int64(len(fetcher.seen)))
}
}
}
// Clear out all fetchers (will crash on a second call, deliberate)
p.fetchers = nil
}
// copy creates a deep-but-inactive copy of the trie prefetcher. Any trie data
// already loaded will be copied over, but no goroutines will be started. This
// is mostly used in the miner which creates a copy of it's actively mutated
// state to be sealed while it may further mutate the state.
func (p *triePrefetcher) copy() *triePrefetcher {
copy := &triePrefetcher{
db: p.db,
root: p.root,
fetches: make(map[string]Trie), // Active prefetchers use the fetches map
deliveryMissMeter: p.deliveryMissMeter,
accountLoadMeter: p.accountLoadMeter,
accountDupMeter: p.accountDupMeter,
accountSkipMeter: p.accountSkipMeter,
accountWasteMeter: p.accountWasteMeter,
storageLoadMeter: p.storageLoadMeter,
storageDupMeter: p.storageDupMeter,
storageSkipMeter: p.storageSkipMeter,
storageWasteMeter: p.storageWasteMeter,
// prefetch schedules a batch of trie items to prefetch. After the prefetcher is
// closed, all the following tasks scheduled will not be executed and an error
// will be returned.
//
// prefetch is called from two locations:
//
// 1. Finalize of the state-objects storage roots. This happens at the end
// of every transaction, meaning that if several transactions touches
// upon the same contract, the parameters invoking this method may be
// repeated.
// 2. Finalize of the main account trie. This happens only once per block.
func (p *triePrefetcher) prefetch(owner common.Hash, root common.Hash, addr common.Address, keys [][]byte) error {
// Ensure the subfetcher is still alive
select {
case <-p.term:
return errTerminated
default:
}
// If the prefetcher is already a copy, duplicate the data
if p.fetches != nil {
for root, fetch := range p.fetches {
if fetch == nil {
continue
}
copy.fetches[root] = p.db.CopyTrie(fetch)
}
return copy
}
// Otherwise we're copying an active fetcher, retrieve the current states
for id, fetcher := range p.fetchers {
copy.fetches[id] = fetcher.peek()
}
return copy
}
// prefetch schedules a batch of trie items to prefetch.
func (p *triePrefetcher) prefetch(owner common.Hash, root common.Hash, addr common.Address, keys [][]byte) {
// If the prefetcher is an inactive one, bail out
if p.fetches != nil {
return
}
// Active fetcher, schedule the retrievals
id := p.trieID(owner, root)
fetcher := p.fetchers[id]
if fetcher == nil {
fetcher = newSubfetcher(p.db, p.root, owner, root, addr)
p.fetchers[id] = fetcher
}
fetcher.schedule(keys)
return fetcher.schedule(keys)
}
// trie returns the trie matching the root hash, or nil if the prefetcher doesn't
// have it.
func (p *triePrefetcher) trie(owner common.Hash, root common.Hash) Trie {
// If the prefetcher is inactive, return from existing deep copies
id := p.trieID(owner, root)
if p.fetches != nil {
trie := p.fetches[id]
if trie == nil {
p.deliveryMissMeter.Mark(1)
return nil
}
return p.db.CopyTrie(trie)
}
// Otherwise the prefetcher is active, bail if no trie was prefetched for this root
fetcher := p.fetchers[id]
// trie returns the trie matching the root hash, blocking until the fetcher of
// the given trie terminates. If no fetcher exists for the request, nil will be
// returned.
func (p *triePrefetcher) trie(owner common.Hash, root common.Hash) (Trie, error) {
// Bail if no trie was prefetched for this root
fetcher := p.fetchers[p.trieID(owner, root)]
if fetcher == nil {
log.Error("Prefetcher missed to load trie", "owner", owner, "root", root)
p.deliveryMissMeter.Mark(1)
return nil
return nil, nil
}
// Interrupt the prefetcher if it's by any chance still running and return
// a copy of any pre-loaded trie.
fetcher.abort() // safe to do multiple times
trie := fetcher.peek()
if trie == nil {
p.deliveryMissMeter.Mark(1)
return nil
}
return trie
// Subfetcher exists, retrieve its trie
return fetcher.peek(), nil
}
// used marks a batch of state items used to allow creating statistics as to
// how useful or wasteful the prefetcher is.
// how useful or wasteful the fetcher is.
func (p *triePrefetcher) used(owner common.Hash, root common.Hash, used [][]byte) {
if fetcher := p.fetchers[p.trieID(owner, root)]; fetcher != nil {
fetcher.wait() // ensure the fetcher's idle before poking in its internals
fetcher.used = used
}
}
@ -221,7 +192,6 @@ type subfetcher struct {
wake chan struct{} // Wake channel if a new task is scheduled
stop chan struct{} // Channel to interrupt processing
term chan struct{} // Channel to signal interruption
copy chan chan Trie // Channel to request a copy of the current trie
seen map[string]struct{} // Tracks the entries already loaded
dups int // Number of duplicate preload tasks
@ -240,7 +210,6 @@ func newSubfetcher(db Database, state common.Hash, owner common.Hash, root commo
wake: make(chan struct{}, 1),
stop: make(chan struct{}),
term: make(chan struct{}),
copy: make(chan chan Trie),
seen: make(map[string]struct{}),
}
go sf.loop()
@ -248,50 +217,61 @@ func newSubfetcher(db Database, state common.Hash, owner common.Hash, root commo
}
// schedule adds a batch of trie keys to the queue to prefetch.
func (sf *subfetcher) schedule(keys [][]byte) {
func (sf *subfetcher) schedule(keys [][]byte) error {
// Ensure the subfetcher is still alive
select {
case <-sf.term:
return errTerminated
default:
}
// Append the tasks to the current queue
sf.lock.Lock()
sf.tasks = append(sf.tasks, keys...)
sf.lock.Unlock()
// Notify the prefetcher, it's fine if it's already terminated
// Notify the background thread to execute scheduled tasks
select {
case sf.wake <- struct{}{}:
// Wake signal sent
default:
// Wake signal not sent as a previous is already queued
}
}
// peek tries to retrieve a deep copy of the fetcher's trie in whatever form it
// is currently.
func (sf *subfetcher) peek() Trie {
ch := make(chan Trie)
select {
case sf.copy <- ch:
// Subfetcher still alive, return copy from it
return <-ch
case <-sf.term:
// Subfetcher already terminated, return a copy directly
if sf.trie == nil {
return nil
}
return sf.db.CopyTrie(sf.trie)
}
}
// abort interrupts the subfetcher immediately. It is safe to call abort multiple
// times but it is not thread safe.
func (sf *subfetcher) abort() {
// wait blocks until the subfetcher terminates. This method is used to block on
// an async termination before accessing internal fields from the fetcher.
func (sf *subfetcher) wait() {
<-sf.term
}
// peek retrieves the fetcher's trie, populated with any pre-fetched data. The
// returned trie will be a shallow copy, so modifying it will break subsequent
// peeks for the original data. The method will block until all the scheduled
// data has been loaded and the fethcer terminated.
func (sf *subfetcher) peek() Trie {
// Block until the fertcher terminates, then retrieve the trie
sf.wait()
return sf.trie
}
// terminate requests the subfetcher to stop accepting new tasks and spin down
// as soon as everything is loaded. Depending on the async parameter, the method
// will either block until all disk loads finish or return immediately.
func (sf *subfetcher) terminate(async bool) {
select {
case <-sf.stop:
default:
close(sf.stop)
}
if async {
return
}
<-sf.term
}
// loop waits for new tasks to be scheduled and keeps loading them until it runs
// out of tasks or its underlying trie is retrieved for committing.
// loop loads newly-scheduled trie tasks as they are received and loads them, stopping
// when requested.
func (sf *subfetcher) loop() {
// No matter how the loop stops, signal anyone waiting that it's terminated
defer close(sf.term)
@ -305,8 +285,6 @@ func (sf *subfetcher) loop() {
}
sf.trie = trie
} else {
// The trie argument can be nil as verkle doesn't support prefetching
// yet. TODO FIX IT(rjl493456442), otherwise code will panic here.
trie, err := sf.db.OpenStorageTrie(sf.state, sf.addr, sf.root, nil)
if err != nil {
log.Warn("Trie prefetcher failed opening trie", "root", sf.root, "err", err)
@ -318,31 +296,17 @@ func (sf *subfetcher) loop() {
for {
select {
case <-sf.wake:
// Subfetcher was woken up, retrieve any tasks to avoid spinning the lock
// Execute all remaining tasks in single run
sf.lock.Lock()
tasks := sf.tasks
sf.tasks = nil
sf.lock.Unlock()
// Prefetch any tasks until the loop is interrupted
for i, task := range tasks {
select {
case <-sf.stop:
// If termination is requested, add any leftover back and return
sf.lock.Lock()
sf.tasks = append(sf.tasks, tasks[i:]...)
sf.lock.Unlock()
return
case ch := <-sf.copy:
// Somebody wants a copy of the current trie, grant them
ch <- sf.db.CopyTrie(sf.trie)
default:
// No termination request yet, prefetch the next entry
for _, task := range tasks {
if _, ok := sf.seen[string(task)]; ok {
sf.dups++
} else {
continue
}
if len(task) == common.AddressLength {
sf.trie.GetAccount(common.BytesToAddress(task))
} else {
@ -350,16 +314,20 @@ func (sf *subfetcher) loop() {
}
sf.seen[string(task)] = struct{}{}
}
}
}
case ch := <-sf.copy:
// Somebody wants a copy of the current trie, grant them
ch <- sf.db.CopyTrie(sf.trie)
case <-sf.stop:
// Termination is requested, abort and leave remaining tasks
// Termination is requested, abort if no more tasks are pending. If
// there are some, exhaust them first.
sf.lock.Lock()
done := sf.tasks == nil
sf.lock.Unlock()
if done {
return
}
// Some tasks are pending, loop and pick them up (that wake branch
// will be selected eventually, whilst stop remains closed to this
// branch will also run afterwards).
}
}
}

65
core/state/trie_prefetcher_test.go
View file

@ -19,7 +19,6 @@ package state
import (
"math/big"
"testing"
"time"
"github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/core/rawdb"
@ -46,68 +45,20 @@ func filledStateDB() *StateDB {
return state
}
func TestCopyAndClose(t *testing.T) {
func TestUseAfterTerminate(t *testing.T) {
db := filledStateDB()
prefetcher := newTriePrefetcher(db.db, db.originalRoot, "")
skey := common.HexToHash("aaa")
prefetcher.prefetch(common.Hash{}, db.originalRoot, common.Address{}, [][]byte{skey.Bytes()})
prefetcher.prefetch(common.Hash{}, db.originalRoot, common.Address{}, [][]byte{skey.Bytes()})
time.Sleep(1 * time.Second)
a := prefetcher.trie(common.Hash{}, db.originalRoot)
prefetcher.prefetch(common.Hash{}, db.originalRoot, common.Address{}, [][]byte{skey.Bytes()})
b := prefetcher.trie(common.Hash{}, db.originalRoot)
cpy := prefetcher.copy()
cpy.prefetch(common.Hash{}, db.originalRoot, common.Address{}, [][]byte{skey.Bytes()})
cpy.prefetch(common.Hash{}, db.originalRoot, common.Address{}, [][]byte{skey.Bytes()})
c := cpy.trie(common.Hash{}, db.originalRoot)
prefetcher.close()
cpy2 := cpy.copy()
cpy2.prefetch(common.Hash{}, db.originalRoot, common.Address{}, [][]byte{skey.Bytes()})
d := cpy2.trie(common.Hash{}, db.originalRoot)
cpy.close()
cpy2.close()
if a.Hash() != b.Hash() || a.Hash() != c.Hash() || a.Hash() != d.Hash() {
t.Fatalf("Invalid trie, hashes should be equal: %v %v %v %v", a.Hash(), b.Hash(), c.Hash(), d.Hash())
}
}
func TestUseAfterClose(t *testing.T) {
db := filledStateDB()
prefetcher := newTriePrefetcher(db.db, db.originalRoot, "")
skey := common.HexToHash("aaa")
prefetcher.prefetch(common.Hash{}, db.originalRoot, common.Address{}, [][]byte{skey.Bytes()})
a := prefetcher.trie(common.Hash{}, db.originalRoot)
prefetcher.close()
b := prefetcher.trie(common.Hash{}, db.originalRoot)
if a == nil {
t.Fatal("Prefetching before close should not return nil")
if err := prefetcher.prefetch(common.Hash{}, db.originalRoot, common.Address{}, [][]byte{skey.Bytes()}); err != nil {
t.Errorf("Prefetch failed before terminate: %v", err)
}
if b != nil {
t.Fatal("Trie after close should return nil")
}
}
prefetcher.terminate(false)
func TestCopyClose(t *testing.T) {
db := filledStateDB()
prefetcher := newTriePrefetcher(db.db, db.originalRoot, "")
skey := common.HexToHash("aaa")
prefetcher.prefetch(common.Hash{}, db.originalRoot, common.Address{}, [][]byte{skey.Bytes()})
cpy := prefetcher.copy()
a := prefetcher.trie(common.Hash{}, db.originalRoot)
b := cpy.trie(common.Hash{}, db.originalRoot)
prefetcher.close()
c := prefetcher.trie(common.Hash{}, db.originalRoot)
d := cpy.trie(common.Hash{}, db.originalRoot)
if a == nil {
t.Fatal("Prefetching before close should not return nil")
if err := prefetcher.prefetch(common.Hash{}, db.originalRoot, common.Address{}, [][]byte{skey.Bytes()}); err == nil {
t.Errorf("Prefetch succeeded after terminate: %v", err)
}
if b == nil {
t.Fatal("Copy trie should return nil")
}
if c != nil {
t.Fatal("Trie after close should return nil")
}
if d == nil {
t.Fatal("Copy trie should not return nil")
if _, err := prefetcher.trie(common.Hash{}, db.originalRoot); err != nil {
t.Errorf("Trie retrieval failed after terminate: %v", err)
}
}

3
core/state_transition.go
View file

@ -240,8 +240,9 @@ func (st *StateTransition) buyGas() error {
if st.msg.GasFeeCap != nil {
balanceCheck.SetUint64(st.msg.GasLimit)
balanceCheck = balanceCheck.Mul(balanceCheck, st.msg.GasFeeCap)
balanceCheck.Add(balanceCheck, st.msg.Value)
}
balanceCheck.Add(balanceCheck, st.msg.Value)
if st.evm.ChainConfig().IsCancun(st.evm.Context.BlockNumber, st.evm.Context.Time) {
if blobGas := st.blobGasUsed(); blobGas > 0 {
// Check that the user has enough funds to cover blobGasUsed * tx.BlobGasFeeCap

2
eth/protocols/snap/sync.go
View file

@ -2358,7 +2358,7 @@ func (s *Syncer) commitHealer(force bool) {
}
batch := s.db.NewBatch()
if err := s.healer.scheduler.Commit(batch); err != nil {
log.Error("Failed to commit healing data", "err", err)
log.Crit("Failed to commit healing data", "err", err)
}
if err := batch.Write(); err != nil {
log.Crit("Failed to persist healing data", "err", err)

8
eth/tracers/api.go
View file

@ -22,7 +22,6 @@ import (
"encoding/json"
"errors"
"fmt"
"math/big"
"os"
"runtime"
"sync"
@ -805,9 +804,13 @@ func (api *API) standardTraceBlockToFile(ctx context.Context, block *types.Block
// Execute the transaction and flush any traces to disk
vmenv := vm.NewEVM(vmctx, txContext, statedb, chainConfig, vmConf)
statedb.SetTxContext(tx.Hash(), i)
if vmConf.Tracer.OnTxStart != nil {
vmConf.Tracer.OnTxStart(vmenv.GetVMContext(), tx, msg.From)
}
vmRet, err := core.ApplyMessage(vmenv, msg, new(core.GasPool).AddGas(msg.GasLimit))
if vmConf.Tracer.OnTxEnd != nil {
vmConf.Tracer.OnTxEnd(&types.Receipt{GasUsed: vmRet.UsedGas}, err)
}
if writer != nil {
writer.Flush()
}
@ -982,7 +985,8 @@ func (api *API) traceTx(ctx context.Context, tx *types.Transaction, message *cor
return nil, err
}
}
vmenv := vm.NewEVM(vmctx, vm.TxContext{GasPrice: big.NewInt(0)}, statedb, api.backend.ChainConfig(), vm.Config{Tracer: tracer.Hooks, NoBaseFee: true})
// The actual TxContext will be created as part of ApplyTransactionWithEVM.
vmenv := vm.NewEVM(vmctx, vm.TxContext{GasPrice: message.GasPrice, BlobFeeCap: message.BlobGasFeeCap}, statedb, api.backend.ChainConfig(), vm.Config{Tracer: tracer.Hooks, NoBaseFee: true})
statedb.SetLogger(tracer.Hooks)
// Define a meaningful timeout of a single transaction trace

88
eth/tracers/api_test.go
View file

@ -32,6 +32,7 @@ import (
"github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/common/hexutil"
"github.com/ethereum/go-ethereum/consensus"
"github.com/ethereum/go-ethereum/consensus/beacon"
"github.com/ethereum/go-ethereum/consensus/ethash"
"github.com/ethereum/go-ethereum/core"
"github.com/ethereum/go-ethereum/core/rawdb"
@ -994,3 +995,90 @@ func TestTraceChain(t *testing.T) {
}
}
}
// newTestMergedBackend creates a post-merge chain
func newTestMergedBackend(t *testing.T, n int, gspec *core.Genesis, generator func(i int, b *core.BlockGen)) *testBackend {
backend := &testBackend{
chainConfig: gspec.Config,
engine: beacon.NewFaker(),
chaindb: rawdb.NewMemoryDatabase(),
}
// Generate blocks for testing
_, blocks, _ := core.GenerateChainWithGenesis(gspec, backend.engine, n, generator)
// Import the canonical chain
cacheConfig := &core.CacheConfig{
TrieCleanLimit: 256,
TrieDirtyLimit: 256,
TrieTimeLimit: 5 * time.Minute,
SnapshotLimit: 0,
TrieDirtyDisabled: true, // Archive mode
}
chain, err := core.NewBlockChain(backend.chaindb, cacheConfig, gspec, nil, backend.engine, vm.Config{}, nil, nil)
if err != nil {
t.Fatalf("failed to create tester chain: %v", err)
}
if n, err := chain.InsertChain(blocks); err != nil {
t.Fatalf("block %d: failed to insert into chain: %v", n, err)
}
backend.chain = chain
return backend
}
func TestTraceBlockWithBasefee(t *testing.T) {
t.Parallel()
accounts := newAccounts(1)
target := common.HexToAddress("0x1111111111111111111111111111111111111111")
genesis := &core.Genesis{
Config: params.AllDevChainProtocolChanges,
Alloc: types.GenesisAlloc{
accounts[0].addr: {Balance: big.NewInt(1 * params.Ether)},
target: {Nonce: 1, Code: []byte{
byte(vm.BASEFEE), byte(vm.STOP),
}},
},
}
genBlocks := 1
signer := types.HomesteadSigner{}
var txHash common.Hash
var baseFee = new(big.Int)
backend := newTestMergedBackend(t, genBlocks, genesis, func(i int, b *core.BlockGen) {
tx, _ := types.SignTx(types.NewTx(&types.LegacyTx{
Nonce: uint64(i),
To: &target,
Value: big.NewInt(0),
Gas: 5 * params.TxGas,
GasPrice: b.BaseFee(),
Data: nil}),
signer, accounts[0].key)
b.AddTx(tx)
txHash = tx.Hash()
baseFee.Set(b.BaseFee())
})
defer backend.chain.Stop()
api := NewAPI(backend)
var testSuite = []struct {
blockNumber rpc.BlockNumber
config *TraceConfig
want string
}{
// Trace head block
{
blockNumber: rpc.BlockNumber(genBlocks),
want: fmt.Sprintf(`[{"txHash":"%#x","result":{"gas":21002,"failed":false,"returnValue":"","structLogs":[{"pc":0,"op":"BASEFEE","gas":84000,"gasCost":2,"depth":1,"stack":[]},{"pc":1,"op":"STOP","gas":83998,"gasCost":0,"depth":1,"stack":["%#x"]}]}}]`, txHash, baseFee),
},
}
for i, tc := range testSuite {
result, err := api.TraceBlockByNumber(context.Background(), tc.blockNumber, tc.config)
if err != nil {
t.Errorf("test %d, want no error, have %v", i, err)
continue
}
have, _ := json.Marshal(result)
want := tc.want
if string(have) != want {
t.Errorf("test %d, result mismatch\nhave: %v\nwant: %v\n", i, string(have), want)
}
}
}

21
eth/tracers/logger/logger_json.go
View file

@ -58,6 +58,7 @@ type jsonLogger struct {
encoder *json.Encoder
cfg *Config
env *tracing.VMContext
hooks *tracing.Hooks
}
// NewJSONLogger creates a new EVM tracer that prints execution steps as JSON objects
@ -67,12 +68,14 @@ func NewJSONLogger(cfg *Config, writer io.Writer) *tracing.Hooks {
if l.cfg == nil {
l.cfg = &Config{}
}
return &tracing.Hooks{
l.hooks = &tracing.Hooks{
OnTxStart: l.OnTxStart,
OnExit: l.OnExit,
OnSystemCallStart: l.onSystemCallStart,
OnExit: l.OnEnd,
OnOpcode: l.OnOpcode,
OnFault: l.OnFault,
}
return l.hooks
}
// NewJSONLoggerWithCallFrames creates a new EVM tracer that prints execution steps as JSON objects
@ -82,13 +85,15 @@ func NewJSONLoggerWithCallFrames(cfg *Config, writer io.Writer) *tracing.Hooks {
if l.cfg == nil {
l.cfg = &Config{}
}
return &tracing.Hooks{
l.hooks = &tracing.Hooks{
OnTxStart: l.OnTxStart,
OnSystemCallStart: l.onSystemCallStart,
OnEnter: l.OnEnter,
OnExit: l.OnExit,
OnOpcode: l.OnOpcode,
OnFault: l.OnFault,
}
return l.hooks
}
func (l *jsonLogger) OnFault(pc uint64, op byte, gas uint64, cost uint64, scope tracing.OpContext, depth int, err error) {
@ -122,6 +127,16 @@ func (l *jsonLogger) OnOpcode(pc uint64, op byte, gas, cost uint64, scope tracin
l.encoder.Encode(log)
}
func (l *jsonLogger) onSystemCallStart() {
// Process no events while in system call.
hooks := *l.hooks
*l.hooks = tracing.Hooks{
OnSystemCallEnd: func() {
*l.hooks = hooks
},
}
}
// OnEnter is not enabled by default.
func (l *jsonLogger) OnEnter(depth int, typ byte, from common.Address, to common.Address, input []byte, gas uint64, value *big.Int) {
frame := callFrame{

4
internal/ethapi/api.go
View file

@ -966,7 +966,7 @@ func (s *BlockChainAPI) GetBlockReceipts(ctx context.Context, blockNrOrHash rpc.
// of a message call.
// Note, state and stateDiff can't be specified at the same time. If state is
// set, message execution will only use the data in the given state. Otherwise
// if statDiff is set, all diff will be applied first and then execute the call
// if stateDiff is set, all diff will be applied first and then execute the call
// message.
type OverrideAccount struct {
Nonce *hexutil.Uint64 `json:"nonce"`
@ -1203,7 +1203,7 @@ func DoEstimateGas(ctx context.Context, b Backend, args TransactionArgs, blockNr
return 0, err
}
call := args.ToMessage(header.BaseFee)
// Run the gas estimation andwrap any revertals into a custom return
// Run the gas estimation and wrap any revertals into a custom return
estimate, revert, err := gasestimator.Estimate(ctx, call, opts, gasCap)
if err != nil {
if len(revert) > 0 {

2
internal/testlog/testlog.go
View file

@ -58,7 +58,7 @@ func (h *bufHandler) Handle(_ context.Context, r slog.Record) error {
}
func (h *bufHandler) Enabled(_ context.Context, lvl slog.Level) bool {
return lvl <= h.level
return lvl >= h.level
}
func (h *bufHandler) WithAttrs(attrs []slog.Attr) slog.Handler {

4
metrics/sample_test.go
View file

@ -114,7 +114,7 @@ func TestExpDecaySample(t *testing.T) {
}
for _, v := range values {
if v > int64(tc.updates) || v < 0 {
t.Errorf("out of range [0, %d): %v", tc.updates, v)
t.Errorf("out of range [0, %d]: %v", tc.updates, v)
}
}
}
@ -195,7 +195,7 @@ func TestUniformSample(t *testing.T) {
}
for _, v := range values {
if v > 1000 || v < 0 {
t.Errorf("out of range [0, 100): %v\n", v)
t.Errorf("out of range [0, 1000]: %v\n", v)
}
}
}

17
node/api.go
View file

@ -26,6 +26,7 @@ import (
"github.com/ethereum/go-ethereum/internal/debug"
"github.com/ethereum/go-ethereum/log"
"github.com/ethereum/go-ethereum/p2p"
"github.com/ethereum/go-ethereum/p2p/discover"
"github.com/ethereum/go-ethereum/p2p/enode"
"github.com/ethereum/go-ethereum/rpc"
)
@ -39,6 +40,9 @@ func (n *Node) apis() []rpc.API {
}, {
Namespace: "debug",
Service: debug.Handler,
}, {
Namespace: "debug",
Service: &p2pDebugAPI{n},
}, {
Namespace: "web3",
Service: &web3API{n},
@ -333,3 +337,16 @@ func (s *web3API) ClientVersion() string {
func (s *web3API) Sha3(input hexutil.Bytes) hexutil.Bytes {
return crypto.Keccak256(input)
}
// p2pDebugAPI provides access to p2p internals for debugging.
type p2pDebugAPI struct {
stack *Node
}
func (s *p2pDebugAPI) DiscoveryV4Table() [][]discover.BucketNode {
disc := s.stack.server.DiscoveryV4()
if disc != nil {
return disc.TableBuckets()
}
return nil
}

47
p2p/discover/common.go
View file

@ -18,7 +18,11 @@ package discover
import (
"crypto/ecdsa"
crand "crypto/rand"
"encoding/binary"
"math/rand"
"net"
"sync"
"time"
"github.com/ethereum/go-ethereum/common/mclock"
@ -62,7 +66,7 @@ type Config struct {
func (cfg Config) withDefaults() Config {
// Node table configuration:
if cfg.PingInterval == 0 {
cfg.PingInterval = 10 * time.Second
cfg.PingInterval = 3 * time.Second
}
if cfg.RefreshInterval == 0 {
cfg.RefreshInterval = 30 * time.Minute
@ -92,3 +96,44 @@ type ReadPacket struct {
Data []byte
Addr *net.UDPAddr
}
type randomSource interface {
Intn(int) int
Int63n(int64) int64
Shuffle(int, func(int, int))
}
// reseedingRandom is a random number generator that tracks when it was last re-seeded.
type reseedingRandom struct {
mu sync.Mutex
cur *rand.Rand
}
func (r *reseedingRandom) seed() {
var b [8]byte
crand.Read(b[:])
seed := binary.BigEndian.Uint64(b[:])
new := rand.New(rand.NewSource(int64(seed)))
r.mu.Lock()
r.cur = new
r.mu.Unlock()
}
func (r *reseedingRandom) Intn(n int) int {
r.mu.Lock()
defer r.mu.Unlock()
return r.cur.Intn(n)
}
func (r *reseedingRandom) Int63n(n int64) int64 {
r.mu.Lock()
defer r.mu.Unlock()
return r.cur.Int63n(n)
}
func (r *reseedingRandom) Shuffle(n int, swap func(i, j int)) {
r.mu.Lock()
defer r.mu.Unlock()
r.cur.Shuffle(n, swap)
}

29
p2p/discover/lookup.go
View file

@ -140,32 +140,13 @@ func (it *lookup) slowdown() {
}
func (it *lookup) query(n *node, reply chan<- []*node) {
fails := it.tab.db.FindFails(n.ID(), n.IP())
r, err := it.queryfunc(n)
if errors.Is(err, errClosed) {
// Avoid recording failures on shutdown.
reply <- nil
return
} else if len(r) == 0 {
fails++
it.tab.db.UpdateFindFails(n.ID(), n.IP(), fails)
// Remove the node from the local table if it fails to return anything useful too
// many times, but only if there are enough other nodes in the bucket.
dropped := false
if fails >= maxFindnodeFailures && it.tab.bucketLen(n.ID()) >= bucketSize/2 {
dropped = true
it.tab.delete(n)
if !errors.Is(err, errClosed) { // avoid recording failures on shutdown.
success := len(r) > 0
it.tab.trackRequest(n, success, r)
if err != nil {
it.tab.log.Trace("FINDNODE failed", "id", n.ID(), "err", err)
}
it.tab.log.Trace("FINDNODE failed", "id", n.ID(), "failcount", fails, "dropped", dropped, "err", err)
} else if fails > 0 {
// Reset failure counter because it counts _consecutive_ failures.
it.tab.db.UpdateFindFails(n.ID(), n.IP(), 0)
}
// Grab as many nodes as possible. Some of them might not be alive anymore, but we'll
// just remove those again during revalidation.
for _, n := range r {
it.tab.addSeenNode(n)
}
reply <- r
}

18
p2p/discover/node.go
View file

@ -29,12 +29,22 @@ import (
"github.com/ethereum/go-ethereum/p2p/enode"
)
type BucketNode struct {
Node *enode.Node `json:"node"`
AddedToTable time.Time `json:"addedToTable"`
AddedToBucket time.Time `json:"addedToBucket"`
Checks int `json:"checks"`
Live bool `json:"live"`
}
// node represents a host on the network.
// The fields of Node may not be modified.
type node struct {
enode.Node
addedAt time.Time // time when the node was added to the table
*enode.Node
addedToTable time.Time // first time node was added to bucket or replacement list
addedToBucket time.Time // time it was added in the actual bucket
livenessChecks uint // how often liveness was checked
isValidatedLive bool // true if existence of node is considered validated right now
}
type encPubkey [64]byte
@ -65,7 +75,7 @@ func (e encPubkey) id() enode.ID {
}
func wrapNode(n *enode.Node) *node {
return &node{Node: *n}
return &node{Node: n}
}
func wrapNodes(ns []*enode.Node) []*node {
@ -77,7 +87,7 @@ func wrapNodes(ns []*enode.Node) []*node {
}
func unwrapNode(n *node) *enode.Node {
return &n.Node
return n.Node
}
func unwrapNodes(ns []*node) []*enode.Node {

653
p2p/discover/table.go
View file

@ -24,16 +24,15 @@ package discover
import (
"context"
crand "crypto/rand"
"encoding/binary"
"fmt"
mrand "math/rand"
"net"
"slices"
"sort"
"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/enode"
@ -55,7 +54,6 @@ const (
bucketIPLimit, bucketSubnet = 2, 24 // at most 2 addresses from the same /24
tableIPLimit, tableSubnet = 10, 24
copyNodesInterval = 30 * time.Second
seedMinTableTime = 5 * time.Minute
seedCount = 30
seedMaxAge = 5 * 24 * time.Hour
@ -68,8 +66,9 @@ type Table struct {
mutex sync.Mutex // protects buckets, bucket content, nursery, rand
buckets [nBuckets]*bucket // index of known nodes by distance
nursery []*node // bootstrap nodes
rand *mrand.Rand // source of randomness, periodically reseeded
rand reseedingRandom // source of randomness, periodically reseeded
ips netutil.DistinctNetSet
revalidation tableRevalidation
db *enode.DB // database of known nodes
net transport
@ -78,6 +77,10 @@ type Table struct {
// loop channels
refreshReq chan chan struct{}
revalResponseCh chan revalidationResponse
addNodeCh chan addNodeOp
addNodeHandled chan bool
trackRequestCh chan trackRequestOp
initDone chan struct{}
closeReq chan struct{}
closed chan struct{}
@ -104,6 +107,17 @@ type bucket struct {
index int
}
type addNodeOp struct {
node *node
isInbound bool
}
type trackRequestOp struct {
node *node
foundNodes []*node
success bool
}
func newTable(t transport, db *enode.DB, cfg Config) (*Table, error) {
cfg = cfg.withDefaults()
tab := &Table{
@ -112,56 +126,49 @@ func newTable(t transport, db *enode.DB, cfg Config) (*Table, error) {
cfg: cfg,
log: cfg.Log,
refreshReq: make(chan chan struct{}),
revalResponseCh: make(chan revalidationResponse),
addNodeCh: make(chan addNodeOp),
addNodeHandled: make(chan bool),
trackRequestCh: make(chan trackRequestOp),
initDone: make(chan struct{}),
closeReq: make(chan struct{}),
closed: make(chan struct{}),
rand: mrand.New(mrand.NewSource(0)),
ips: netutil.DistinctNetSet{Subnet: tableSubnet, Limit: tableIPLimit},
}
if err := tab.setFallbackNodes(cfg.Bootnodes); err != nil {
return nil, err
}
for i := range tab.buckets {
tab.buckets[i] = &bucket{
index: i,
ips: netutil.DistinctNetSet{Subnet: bucketSubnet, Limit: bucketIPLimit},
}
}
tab.seedRand()
tab.rand.seed()
tab.revalidation.init(&cfg)
// initial table content
if err := tab.setFallbackNodes(cfg.Bootnodes); err != nil {
return nil, err
}
tab.loadSeedNodes()
return tab, nil
}
func newMeteredTable(t transport, db *enode.DB, cfg Config) (*Table, error) {
tab, err := newTable(t, db, cfg)
if err != nil {
return nil, err
}
if metrics.Enabled {
tab.nodeAddedHook = func(b *bucket, n *node) {
bucketsCounter[b.index].Inc(1)
}
tab.nodeRemovedHook = func(b *bucket, n *node) {
bucketsCounter[b.index].Dec(1)
}
}
return tab, nil
}
// Nodes returns all nodes contained in the table.
func (tab *Table) Nodes() []*enode.Node {
if !tab.isInitDone() {
return nil
}
func (tab *Table) Nodes() [][]BucketNode {
tab.mutex.Lock()
defer tab.mutex.Unlock()
var nodes []*enode.Node
for _, b := range &tab.buckets {
for _, n := range b.entries {
nodes = append(nodes, unwrapNode(n))
nodes := make([][]BucketNode, len(tab.buckets))
for i, b := range &tab.buckets {
nodes[i] = make([]BucketNode, len(b.entries))
for j, n := range b.entries {
nodes[i][j] = BucketNode{
Node: n.Node,
Checks: int(n.livenessChecks),
Live: n.isValidatedLive,
AddedToTable: n.addedToTable,
AddedToBucket: n.addedToBucket,
}
}
}
return nodes
@ -171,15 +178,6 @@ func (tab *Table) self() *enode.Node {
return tab.net.Self()
}
func (tab *Table) seedRand() {
var b [8]byte
crand.Read(b[:])
tab.mutex.Lock()
tab.rand.Seed(int64(binary.BigEndian.Uint64(b[:])))
tab.mutex.Unlock()
}
// getNode returns the node with the given ID or nil if it isn't in the table.
func (tab *Table) getNode(id enode.ID) *enode.Node {
tab.mutex.Lock()
@ -239,52 +237,173 @@ func (tab *Table) refresh() <-chan struct{} {
return done
}
// loop schedules runs of doRefresh, doRevalidate and copyLiveNodes.
// findnodeByID returns the n nodes in the table that are closest to the given id.
// This is used by the FINDNODE/v4 handler.
//
// The preferLive parameter says whether the caller wants liveness-checked results. If
// preferLive is true and the table contains any verified nodes, the result will not
// contain unverified nodes. However, if there are no verified nodes at all, the result
// will contain unverified nodes.
func (tab *Table) findnodeByID(target enode.ID, nresults int, preferLive bool) *nodesByDistance {
tab.mutex.Lock()
defer tab.mutex.Unlock()
// Scan all buckets. There might be a better way to do this, but there aren't that many
// buckets, so this solution should be fine. The worst-case complexity of this loop
// is O(tab.len() * nresults).
nodes := &nodesByDistance{target: target}
liveNodes := &nodesByDistance{target: target}
for _, b := range &tab.buckets {
for _, n := range b.entries {
nodes.push(n, nresults)
if preferLive && n.isValidatedLive {
liveNodes.push(n, nresults)
}
}
}
if preferLive && len(liveNodes.entries) > 0 {
return liveNodes
}
return nodes
}
// appendLiveNodes adds nodes at the given distance to the result slice.
// This is used by the FINDNODE/v5 handler.
func (tab *Table) appendLiveNodes(dist uint, result []*enode.Node) []*enode.Node {
if dist > 256 {
return result
}
if dist == 0 {
return append(result, tab.self())
}
tab.mutex.Lock()
for _, n := range tab.bucketAtDistance(int(dist)).entries {
if n.isValidatedLive {
result = append(result, n.Node)
}
}
tab.mutex.Unlock()
// Shuffle result to avoid always returning same nodes in FINDNODE/v5.
tab.rand.Shuffle(len(result), func(i, j int) {
result[i], result[j] = result[j], result[i]
})
return result
}
// len returns the number of nodes in the table.
func (tab *Table) len() (n int) {
tab.mutex.Lock()
defer tab.mutex.Unlock()
for _, b := range &tab.buckets {
n += len(b.entries)
}
return n
}
// addFoundNode adds a node which may not be live. If the bucket has space available,
// adding the node succeeds immediately. Otherwise, the node is added to the replacements
// list.
//
// The caller must not hold tab.mutex.
func (tab *Table) addFoundNode(n *node) bool {
op := addNodeOp{node: n, isInbound: false}
select {
case tab.addNodeCh <- op:
return <-tab.addNodeHandled
case <-tab.closeReq:
return false
}
}
// addInboundNode adds a node from an inbound contact. If the bucket has no space, the
// node is added to the replacements list.
//
// There is an additional safety measure: if the table is still initializing the node is
// not added. This prevents an attack where the table could be filled by just sending ping
// repeatedly.
//
// The caller must not hold tab.mutex.
func (tab *Table) addInboundNode(n *node) bool {
op := addNodeOp{node: n, isInbound: true}
select {
case tab.addNodeCh <- op:
return <-tab.addNodeHandled
case <-tab.closeReq:
return false
}
}
func (tab *Table) trackRequest(n *node, success bool, foundNodes []*node) {
op := trackRequestOp{n, foundNodes, success}
select {
case tab.trackRequestCh <- op:
case <-tab.closeReq:
}
}
// loop is the main loop of Table.
func (tab *Table) loop() {
var (
revalidate = time.NewTimer(tab.nextRevalidateTime())
refresh = time.NewTimer(tab.nextRefreshTime())
copyNodes = time.NewTicker(copyNodesInterval)
refreshDone = make(chan struct{}) // where doRefresh reports completion
revalidateDone chan struct{} // where doRevalidate reports completion
waiting = []chan struct{}{tab.initDone} // holds waiting callers while doRefresh runs
revalTimer = mclock.NewAlarm(tab.cfg.Clock)
reseedRandTimer = time.NewTicker(10 * time.Minute)
)
defer refresh.Stop()
defer revalidate.Stop()
defer copyNodes.Stop()
defer revalTimer.Stop()
defer reseedRandTimer.Stop()
// Start initial refresh.
go tab.doRefresh(refreshDone)
loop:
for {
nextTime := tab.revalidation.run(tab, tab.cfg.Clock.Now())
revalTimer.Schedule(nextTime)
select {
case <-reseedRandTimer.C:
tab.rand.seed()
case <-revalTimer.C():
case r := <-tab.revalResponseCh:
tab.revalidation.handleResponse(tab, r)
case op := <-tab.addNodeCh:
tab.mutex.Lock()
ok := tab.handleAddNode(op)
tab.mutex.Unlock()
tab.addNodeHandled <- ok
case op := <-tab.trackRequestCh:
tab.handleTrackRequest(op)
case <-refresh.C:
tab.seedRand()
if refreshDone == nil {
refreshDone = make(chan struct{})
go tab.doRefresh(refreshDone)
}
case req := <-tab.refreshReq:
waiting = append(waiting, req)
if refreshDone == nil {
refreshDone = make(chan struct{})
go tab.doRefresh(refreshDone)
}
case <-refreshDone:
for _, ch := range waiting {
close(ch)
}
waiting, refreshDone = nil, nil
refresh.Reset(tab.nextRefreshTime())
case <-revalidate.C:
revalidateDone = make(chan struct{})
go tab.doRevalidate(revalidateDone)
case <-revalidateDone:
revalidate.Reset(tab.nextRevalidateTime())
revalidateDone = nil
case <-copyNodes.C:
go tab.copyLiveNodes()
case <-tab.closeReq:
break loop
}
@ -296,9 +415,6 @@ loop:
for _, ch := range waiting {
close(ch)
}
if revalidateDone != nil {
<-revalidateDone
}
close(tab.closed)
}
@ -335,169 +451,15 @@ func (tab *Table) loadSeedNodes() {
age := time.Since(tab.db.LastPongReceived(seed.ID(), seed.IP()))
tab.log.Trace("Found seed node in database", "id", seed.ID(), "addr", seed.addr(), "age", age)
}
tab.addSeenNode(seed)
tab.handleAddNode(addNodeOp{node: seed, isInbound: false})
}
}
// doRevalidate checks that the last node in a random bucket is still live and replaces or
// deletes the node if it isn't.
func (tab *Table) doRevalidate(done chan<- struct{}) {
defer func() { done <- struct{}{} }()
last, bi := tab.nodeToRevalidate()
if last == nil {
// No non-empty bucket found.
return
}
// Ping the selected node and wait for a pong.
remoteSeq, err := tab.net.ping(unwrapNode(last))
// Also fetch record if the node replied and returned a higher sequence number.
if last.Seq() < remoteSeq {
n, err := tab.net.RequestENR(unwrapNode(last))
if err != nil {
tab.log.Debug("ENR request failed", "id", last.ID(), "addr", last.addr(), "err", err)
} else {
last = &node{Node: *n, addedAt: last.addedAt, livenessChecks: last.livenessChecks}
}
}
tab.mutex.Lock()
defer tab.mutex.Unlock()
b := tab.buckets[bi]
if err == nil {
// The node responded, move it to the front.
last.livenessChecks++
tab.log.Debug("Revalidated node", "b", bi, "id", last.ID(), "checks", last.livenessChecks)
tab.bumpInBucket(b, last)
return
}
// No reply received, pick a replacement or delete the node if there aren't
// any replacements.
if r := tab.replace(b, last); r != nil {
tab.log.Debug("Replaced dead node", "b", bi, "id", last.ID(), "ip", last.IP(), "checks", last.livenessChecks, "r", r.ID(), "rip", r.IP())
} else {
tab.log.Debug("Removed dead node", "b", bi, "id", last.ID(), "ip", last.IP(), "checks", last.livenessChecks)
}
}
// nodeToRevalidate returns the last node in a random, non-empty bucket.
func (tab *Table) nodeToRevalidate() (n *node, bi int) {
tab.mutex.Lock()
defer tab.mutex.Unlock()
for _, bi = range tab.rand.Perm(len(tab.buckets)) {
b := tab.buckets[bi]
if len(b.entries) > 0 {
last := b.entries[len(b.entries)-1]
return last, bi
}
}
return nil, 0
}
func (tab *Table) nextRevalidateTime() time.Duration {
tab.mutex.Lock()
defer tab.mutex.Unlock()
return time.Duration(tab.rand.Int63n(int64(tab.cfg.PingInterval)))
}
func (tab *Table) nextRefreshTime() time.Duration {
tab.mutex.Lock()
defer tab.mutex.Unlock()
half := tab.cfg.RefreshInterval / 2
return half + time.Duration(tab.rand.Int63n(int64(half)))
}
// copyLiveNodes adds nodes from the table to the database if they have been in the table
// longer than seedMinTableTime.
func (tab *Table) copyLiveNodes() {
tab.mutex.Lock()
defer tab.mutex.Unlock()
now := time.Now()
for _, b := range &tab.buckets {
for _, n := range b.entries {
if n.livenessChecks > 0 && now.Sub(n.addedAt) >= seedMinTableTime {
tab.db.UpdateNode(unwrapNode(n))
}
}
}
}
// findnodeByID returns the n nodes in the table that are closest to the given id.
// This is used by the FINDNODE/v4 handler.
//
// The preferLive parameter says whether the caller wants liveness-checked results. If
// preferLive is true and the table contains any verified nodes, the result will not
// contain unverified nodes. However, if there are no verified nodes at all, the result
// will contain unverified nodes.
func (tab *Table) findnodeByID(target enode.ID, nresults int, preferLive bool) *nodesByDistance {
tab.mutex.Lock()
defer tab.mutex.Unlock()
// Scan all buckets. There might be a better way to do this, but there aren't that many
// buckets, so this solution should be fine. The worst-case complexity of this loop
// is O(tab.len() * nresults).
nodes := &nodesByDistance{target: target}
liveNodes := &nodesByDistance{target: target}
for _, b := range &tab.buckets {
for _, n := range b.entries {
nodes.push(n, nresults)
if preferLive && n.livenessChecks > 0 {
liveNodes.push(n, nresults)
}
}
}
if preferLive && len(liveNodes.entries) > 0 {
return liveNodes
}
return nodes
}
// appendLiveNodes adds nodes at the given distance to the result slice.
func (tab *Table) appendLiveNodes(dist uint, result []*enode.Node) []*enode.Node {
if dist > 256 {
return result
}
if dist == 0 {
return append(result, tab.self())
}
tab.mutex.Lock()
defer tab.mutex.Unlock()
for _, n := range tab.bucketAtDistance(int(dist)).entries {
if n.livenessChecks >= 1 {
node := n.Node // avoid handing out pointer to struct field
result = append(result, &node)
}
}
return result
}
// len returns the number of nodes in the table.
func (tab *Table) len() (n int) {
tab.mutex.Lock()
defer tab.mutex.Unlock()
for _, b := range &tab.buckets {
n += len(b.entries)
}
return n
}
// bucketLen returns the number of nodes in the bucket for the given ID.
func (tab *Table) bucketLen(id enode.ID) int {
tab.mutex.Lock()
defer tab.mutex.Unlock()
return len(tab.bucket(id).entries)
}
// bucket returns the bucket for the given node ID hash.
func (tab *Table) bucket(id enode.ID) *bucket {
d := enode.LogDist(tab.self().ID(), id)
@ -511,95 +473,6 @@ func (tab *Table) bucketAtDistance(d int) *bucket {
return tab.buckets[d-bucketMinDistance-1]
}
// addSeenNode adds a node which may or may not be live to the end of a bucket. If the
// bucket has space available, adding the node succeeds immediately. Otherwise, the node is
// added to the replacements list.
//
// The caller must not hold tab.mutex.
func (tab *Table) addSeenNode(n *node) {
if n.ID() == tab.self().ID() {
return
}
tab.mutex.Lock()
defer tab.mutex.Unlock()
b := tab.bucket(n.ID())
if contains(b.entries, n.ID()) {
// Already in bucket, don't add.
return
}
if len(b.entries) >= bucketSize {
// Bucket full, maybe add as replacement.
tab.addReplacement(b, n)
return
}
if !tab.addIP(b, n.IP()) {
// Can't add: IP limit reached.
return
}
// Add to end of bucket:
b.entries = append(b.entries, n)
b.replacements = deleteNode(b.replacements, n)
n.addedAt = time.Now()
if tab.nodeAddedHook != nil {
tab.nodeAddedHook(b, n)
}
}
// addVerifiedNode adds a node whose existence has been verified recently to the front of a
// bucket. If the node is already in the bucket, it is moved to the front. If the bucket
// has no space, the node is added to the replacements list.
//
// There is an additional safety measure: if the table is still initializing the node
// is not added. This prevents an attack where the table could be filled by just sending
// ping repeatedly.
//
// The caller must not hold tab.mutex.
func (tab *Table) addVerifiedNode(n *node) {
if !tab.isInitDone() {
return
}
if n.ID() == tab.self().ID() {
return
}
tab.mutex.Lock()
defer tab.mutex.Unlock()
b := tab.bucket(n.ID())
if tab.bumpInBucket(b, n) {
// Already in bucket, moved to front.
return
}
if len(b.entries) >= bucketSize {
// Bucket full, maybe add as replacement.
tab.addReplacement(b, n)
return
}
if !tab.addIP(b, n.IP()) {
// Can't add: IP limit reached.
return
}
// Add to front of bucket.
b.entries, _ = pushNode(b.entries, n, bucketSize)
b.replacements = deleteNode(b.replacements, n)
n.addedAt = time.Now()
if tab.nodeAddedHook != nil {
tab.nodeAddedHook(b, n)
}
}
// delete removes an entry from the node table. It is used to evacuate dead nodes.
func (tab *Table) delete(node *node) {
tab.mutex.Lock()
defer tab.mutex.Unlock()
tab.deleteInBucket(tab.bucket(node.ID()), node)
}
func (tab *Table) addIP(b *bucket, ip net.IP) bool {
if len(ip) == 0 {
return false // Nodes without IP cannot be added.
@ -627,15 +500,51 @@ func (tab *Table) removeIP(b *bucket, ip net.IP) {
b.ips.Remove(ip)
}
func (tab *Table) addReplacement(b *bucket, n *node) {
for _, e := range b.replacements {
if e.ID() == n.ID() {
return // already in list
// handleAddNode adds the node in the request to the table, if there is space.
// The caller must hold tab.mutex.
func (tab *Table) handleAddNode(req addNodeOp) bool {
if req.node.ID() == tab.self().ID() {
return false
}
// For nodes from inbound contact, there is an additional safety measure: if the table
// is still initializing the node is not added.
if req.isInbound && !tab.isInitDone() {
return false
}
b := tab.bucket(req.node.ID())
if tab.bumpInBucket(b, req.node.Node) {
// Already in bucket, update record.
return false
}
if len(b.entries) >= bucketSize {
// Bucket full, maybe add as replacement.
tab.addReplacement(b, req.node)
return false
}
if !tab.addIP(b, req.node.IP()) {
// Can't add: IP limit reached.
return false
}
// Add to bucket.
b.entries = append(b.entries, req.node)
b.replacements = deleteNode(b.replacements, req.node)
tab.nodeAdded(b, req.node)
return true
}
// addReplacement adds n to the replacement cache of bucket b.
func (tab *Table) addReplacement(b *bucket, n *node) {
if contains(b.replacements, n.ID()) {
// TODO: update ENR
return
}
if !tab.addIP(b, n.IP()) {
return
}
n.addedToTable = time.Now()
var removed *node
b.replacements, removed = pushNode(b.replacements, n, maxReplacements)
if removed != nil {
@ -643,59 +552,107 @@ func (tab *Table) addReplacement(b *bucket, n *node) {
}
}
// replace removes n from the replacement list and replaces 'last' with it if it is the
// last entry in the bucket. If 'last' isn't the last entry, it has either been replaced
// with someone else or became active.
func (tab *Table) replace(b *bucket, last *node) *node {
if len(b.entries) == 0 || b.entries[len(b.entries)-1].ID() != last.ID() {
// Entry has moved, don't replace it.
return nil
func (tab *Table) nodeAdded(b *bucket, n *node) {
if n.addedToTable == (time.Time{}) {
n.addedToTable = time.Now()
}
// Still the last entry.
if len(b.replacements) == 0 {
tab.deleteInBucket(b, last)
return nil
n.addedToBucket = time.Now()
tab.revalidation.nodeAdded(tab, n)
if tab.nodeAddedHook != nil {
tab.nodeAddedHook(b, n)
}
if metrics.Enabled {
bucketsCounter[b.index].Inc(1)
}
r := b.replacements[tab.rand.Intn(len(b.replacements))]
b.replacements = deleteNode(b.replacements, r)
b.entries[len(b.entries)-1] = r
tab.removeIP(b, last.IP())
return r
}
// bumpInBucket moves the given node to the front of the bucket entry list
// if it is contained in that list.
func (tab *Table) bumpInBucket(b *bucket, n *node) bool {
for i := range b.entries {
if b.entries[i].ID() == n.ID() {
if !n.IP().Equal(b.entries[i].IP()) {
func (tab *Table) nodeRemoved(b *bucket, n *node) {
tab.revalidation.nodeRemoved(n)
if tab.nodeRemovedHook != nil {
tab.nodeRemovedHook(b, n)
}
if metrics.Enabled {
bucketsCounter[b.index].Dec(1)
}
}
// deleteInBucket removes node n from the table.
// If there are replacement nodes in the bucket, the node is replaced.
func (tab *Table) deleteInBucket(b *bucket, id enode.ID) *node {
index := slices.IndexFunc(b.entries, func(e *node) bool { return e.ID() == id })
if index == -1 {
// Entry has been removed already.
return nil
}
// Remove the node.
n := b.entries[index]
b.entries = slices.Delete(b.entries, index, index+1)
tab.removeIP(b, n.IP())
tab.nodeRemoved(b, n)
// Add replacement.
if len(b.replacements) == 0 {
tab.log.Debug("Removed dead node", "b", b.index, "id", n.ID(), "ip", n.IP())
return nil
}
rindex := tab.rand.Intn(len(b.replacements))
rep := b.replacements[rindex]
b.replacements = slices.Delete(b.replacements, rindex, rindex+1)
b.entries = append(b.entries, rep)
tab.nodeAdded(b, rep)
tab.log.Debug("Replaced dead node", "b", b.index, "id", n.ID(), "ip", n.IP(), "r", rep.ID(), "rip", rep.IP())
return rep
}
// bumpInBucket updates the node record of n in the bucket.
func (tab *Table) bumpInBucket(b *bucket, newRecord *enode.Node) bool {
i := slices.IndexFunc(b.entries, func(elem *node) bool {
return elem.ID() == newRecord.ID()
})
if i == -1 {
return false
}
if !newRecord.IP().Equal(b.entries[i].IP()) {
// Endpoint has changed, ensure that the new IP fits into table limits.
tab.removeIP(b, b.entries[i].IP())
if !tab.addIP(b, n.IP()) {
if !tab.addIP(b, newRecord.IP()) {
// It doesn't, put the previous one back.
tab.addIP(b, b.entries[i].IP())
return false
}
}
// Move it to the front.
copy(b.entries[1:], b.entries[:i])
b.entries[0] = n
b.entries[i].Node = newRecord
return true
}
}
return false
}
func (tab *Table) deleteInBucket(b *bucket, n *node) {
// Check if the node is actually in the bucket so the removed hook
// isn't called multiple times for the same node.
if !contains(b.entries, n.ID()) {
return
func (tab *Table) handleTrackRequest(op trackRequestOp) {
var fails int
if op.success {
// Reset failure counter because it counts _consecutive_ failures.
tab.db.UpdateFindFails(op.node.ID(), op.node.IP(), 0)
} else {
fails = tab.db.FindFails(op.node.ID(), op.node.IP())
fails++
tab.db.UpdateFindFails(op.node.ID(), op.node.IP(), fails)
}
b.entries = deleteNode(b.entries, n)
tab.removeIP(b, n.IP())
if tab.nodeRemovedHook != nil {
tab.nodeRemovedHook(b, n)
tab.mutex.Lock()
defer tab.mutex.Unlock()
b := tab.bucket(op.node.ID())
// Remove the node from the local table if it fails to return anything useful too
// many times, but only if there are enough other nodes in the bucket. This latter
// condition specifically exists to make bootstrapping in smaller test networks more
// reliable.
if fails >= maxFindnodeFailures && len(b.entries) >= bucketSize/4 {
tab.deleteInBucket(b, op.node.ID())
}
// Add found nodes.
for _, n := range op.foundNodes {
tab.handleAddNode(addNodeOp{n, false})
}
}

223
p2p/discover/table_reval.go Normal file
View file

@ -0,0 +1,223 @@
// Copyright 2024 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 discover
import (
"fmt"
"math"
"slices"
"time"
"github.com/ethereum/go-ethereum/common/mclock"
"github.com/ethereum/go-ethereum/p2p/enode"
)
const never = mclock.AbsTime(math.MaxInt64)
// tableRevalidation implements the node revalidation process.
// It tracks all nodes contained in Table, and schedules sending PING to them.
type tableRevalidation struct {
fast revalidationList
slow revalidationList
activeReq map[enode.ID]struct{}
}
type revalidationResponse struct {
n *node
newRecord *enode.Node
list *revalidationList
didRespond bool
}
func (tr *tableRevalidation) init(cfg *Config) {
tr.activeReq = make(map[enode.ID]struct{})
tr.fast.nextTime = never
tr.fast.interval = cfg.PingInterval
tr.fast.name = "fast"
tr.slow.nextTime = never
tr.slow.interval = cfg.PingInterval * 3
tr.slow.name = "slow"
}
// nodeAdded is called when the table receives a new node.
func (tr *tableRevalidation) nodeAdded(tab *Table, n *node) {
tr.fast.push(n, tab.cfg.Clock.Now(), &tab.rand)
}
// nodeRemoved is called when a node was removed from the table.
func (tr *tableRevalidation) nodeRemoved(n *node) {
if !tr.fast.remove(n) {
tr.slow.remove(n)
}
}
// run performs node revalidation.
// It returns the next time it should be invoked, which is used in the Table main loop
// to schedule a timer. However, run can be called at any time.
func (tr *tableRevalidation) run(tab *Table, now mclock.AbsTime) (nextTime mclock.AbsTime) {
if n := tr.fast.get(now, &tab.rand, tr.activeReq); n != nil {
tr.startRequest(tab, &tr.fast, n)
tr.fast.schedule(now, &tab.rand)
}
if n := tr.slow.get(now, &tab.rand, tr.activeReq); n != nil {
tr.startRequest(tab, &tr.slow, n)
tr.slow.schedule(now, &tab.rand)
}
return min(tr.fast.nextTime, tr.slow.nextTime)
}
// startRequest spawns a revalidation request for node n.
func (tr *tableRevalidation) startRequest(tab *Table, list *revalidationList, n *node) {
if _, ok := tr.activeReq[n.ID()]; ok {
panic(fmt.Errorf("duplicate startRequest (list %q, node %v)", list.name, n.ID()))
}
tr.activeReq[n.ID()] = struct{}{}
resp := revalidationResponse{n: n, list: list}
// Fetch the node while holding lock.
tab.mutex.Lock()
node := n.Node
tab.mutex.Unlock()
go tab.doRevalidate(resp, node)
}
func (tab *Table) doRevalidate(resp revalidationResponse, node *enode.Node) {
// Ping the selected node and wait for a pong response.
remoteSeq, err := tab.net.ping(node)
resp.didRespond = err == nil
// Also fetch record if the node replied and returned a higher sequence number.
if remoteSeq > node.Seq() {
newrec, err := tab.net.RequestENR(node)
if err != nil {
tab.log.Debug("ENR request failed", "id", node.ID(), "err", err)
} else {
resp.newRecord = newrec
}
}
select {
case tab.revalResponseCh <- resp:
case <-tab.closed:
}
}
// handleResponse processes the result of a revalidation request.
func (tr *tableRevalidation) handleResponse(tab *Table, resp revalidationResponse) {
now := tab.cfg.Clock.Now()
n := resp.n
b := tab.bucket(n.ID())
delete(tr.activeReq, n.ID())
tab.mutex.Lock()
defer tab.mutex.Unlock()
if !resp.didRespond {
// Revalidation failed.
n.livenessChecks /= 3
if n.livenessChecks <= 0 {
tab.deleteInBucket(b, n.ID())
} else {
tr.moveToList(&tr.fast, resp.list, n, now, &tab.rand)
}
return
}
// The node responded.
n.livenessChecks++
n.isValidatedLive = true
var endpointChanged bool
if resp.newRecord != nil {
endpointChanged = tab.bumpInBucket(b, resp.newRecord)
if endpointChanged {
// If the node changed its advertised endpoint, the updated ENR is not served
// until it has been revalidated.
n.isValidatedLive = false
}
}
tab.log.Debug("Revalidated node", "b", b.index, "id", n.ID(), "checks", n.livenessChecks, "q", resp.list.name)
// Move node over to slow queue after first validation.
if !endpointChanged {
tr.moveToList(&tr.slow, resp.list, n, now, &tab.rand)
} else {
tr.moveToList(&tr.fast, resp.list, n, now, &tab.rand)
}
// Store potential seeds in database.
if n.isValidatedLive && n.livenessChecks > 5 {
tab.db.UpdateNode(resp.n.Node)
}
}
func (tr *tableRevalidation) moveToList(dest, source *revalidationList, n *node, now mclock.AbsTime, rand randomSource) {
if source == dest {
return
}
if !source.remove(n) {
panic(fmt.Errorf("moveToList(%q -> %q): node %v not in source list", source.name, dest.name, n.ID()))
}
dest.push(n, now, rand)
}
// revalidationList holds a list nodes and the next revalidation time.
type revalidationList struct {
nodes []*node
nextTime mclock.AbsTime
interval time.Duration
name string
}
// get returns a random node from the queue. Nodes in the 'exclude' map are not returned.
func (list *revalidationList) get(now mclock.AbsTime, rand randomSource, exclude map[enode.ID]struct{}) *node {
if now < list.nextTime || len(list.nodes) == 0 {
return nil
}
for i := 0; i < len(list.nodes)*3; i++ {
n := list.nodes[rand.Intn(len(list.nodes))]
_, excluded := exclude[n.ID()]
if !excluded {
return n
}
}
return nil
}
func (list *revalidationList) schedule(now mclock.AbsTime, rand randomSource) {
list.nextTime = now.Add(time.Duration(rand.Int63n(int64(list.interval))))
}
func (list *revalidationList) push(n *node, now mclock.AbsTime, rand randomSource) {
list.nodes = append(list.nodes, n)
if list.nextTime == never {
list.schedule(now, rand)
}
}
func (list *revalidationList) remove(n *node) bool {
i := slices.Index(list.nodes, n)
if i == -1 {
return false
}
list.nodes = slices.Delete(list.nodes, i, i+1)
if len(list.nodes) == 0 {
list.nextTime = never
}
return true
}

177
p2p/discover/table_test.go
View file

@ -20,14 +20,16 @@ import (
"crypto/ecdsa"
"fmt"
"math/rand"
"net"
"reflect"
"testing"
"testing/quick"
"time"
"github.com/ethereum/go-ethereum/common/mclock"
"github.com/ethereum/go-ethereum/crypto"
"github.com/ethereum/go-ethereum/internal/testlog"
"github.com/ethereum/go-ethereum/log"
"github.com/ethereum/go-ethereum/p2p/enode"
"github.com/ethereum/go-ethereum/p2p/enr"
"github.com/ethereum/go-ethereum/p2p/netutil"
@ -49,106 +51,109 @@ func TestTable_pingReplace(t *testing.T) {
}
func testPingReplace(t *testing.T, newNodeIsResponding, lastInBucketIsResponding bool) {
simclock := new(mclock.Simulated)
transport := newPingRecorder()
tab, db := newTestTable(transport)
tab, db := newTestTable(transport, Config{
Clock: simclock,
Log: testlog.Logger(t, log.LevelTrace),
})
defer db.Close()
defer tab.close()
<-tab.initDone
// Fill up the sender's bucket.
pingKey, _ := crypto.HexToECDSA("45a915e4d060149eb4365960e6a7a45f334393093061116b197e3240065ff2d8")
pingSender := wrapNode(enode.NewV4(&pingKey.PublicKey, net.IP{127, 0, 0, 1}, 99, 99))
last := fillBucket(tab, pingSender)
replacementNodeKey, _ := crypto.HexToECDSA("45a915e4d060149eb4365960e6a7a45f334393093061116b197e3240065ff2d8")
replacementNode := wrapNode(enode.NewV4(&replacementNodeKey.PublicKey, net.IP{127, 0, 0, 1}, 99, 99))
last := fillBucket(tab, replacementNode.ID())
tab.mutex.Lock()
nodeEvents := newNodeEventRecorder(128)
tab.nodeAddedHook = nodeEvents.nodeAdded
tab.nodeRemovedHook = nodeEvents.nodeRemoved
tab.mutex.Unlock()
// Add the sender as if it just pinged us. Revalidate should replace the last node in
// its bucket if it is unresponsive. Revalidate again to ensure that
// The revalidation process should replace
// this node in the bucket if it is unresponsive.
transport.dead[last.ID()] = !lastInBucketIsResponding
transport.dead[pingSender.ID()] = !newNodeIsResponding
tab.addSeenNode(pingSender)
tab.doRevalidate(make(chan struct{}, 1))
tab.doRevalidate(make(chan struct{}, 1))
transport.dead[replacementNode.ID()] = !newNodeIsResponding
if !transport.pinged[last.ID()] {
// Oldest node in bucket is pinged to see whether it is still alive.
t.Error("table did not ping last node in bucket")
// Add replacement node to table.
tab.addFoundNode(replacementNode)
t.Log("last:", last.ID())
t.Log("replacement:", replacementNode.ID())
// Wait until the last node was pinged.
waitForRevalidationPing(t, transport, tab, last.ID())
if !lastInBucketIsResponding {
if !nodeEvents.waitNodeAbsent(last.ID(), 2*time.Second) {
t.Error("last node was not removed")
}
if !nodeEvents.waitNodePresent(replacementNode.ID(), 2*time.Second) {
t.Error("replacement node was not added")
}
// If a replacement is expected, we also need to wait until the replacement node
// was pinged and added/removed.
waitForRevalidationPing(t, transport, tab, replacementNode.ID())
if !newNodeIsResponding {
if !nodeEvents.waitNodeAbsent(replacementNode.ID(), 2*time.Second) {
t.Error("replacement node was not removed")
}
}
}
// Check bucket content.
tab.mutex.Lock()
defer tab.mutex.Unlock()
wantSize := bucketSize
if !lastInBucketIsResponding && !newNodeIsResponding {
wantSize--
}
if l := len(tab.bucket(pingSender.ID()).entries); l != wantSize {
t.Errorf("wrong bucket size after bond: got %d, want %d", l, wantSize)
bucket := tab.bucket(replacementNode.ID())
if l := len(bucket.entries); l != wantSize {
t.Errorf("wrong bucket size after revalidation: got %d, want %d", l, wantSize)
}
if found := contains(tab.bucket(pingSender.ID()).entries, last.ID()); found != lastInBucketIsResponding {
t.Errorf("last entry found: %t, want: %t", found, lastInBucketIsResponding)
if ok := contains(bucket.entries, last.ID()); ok != lastInBucketIsResponding {
t.Errorf("revalidated node found: %t, want: %t", ok, lastInBucketIsResponding)
}
wantNewEntry := newNodeIsResponding && !lastInBucketIsResponding
if found := contains(tab.bucket(pingSender.ID()).entries, pingSender.ID()); found != wantNewEntry {
t.Errorf("new entry found: %t, want: %t", found, wantNewEntry)
if ok := contains(bucket.entries, replacementNode.ID()); ok != wantNewEntry {
t.Errorf("replacement node found: %t, want: %t", ok, wantNewEntry)
}
}
func TestBucket_bumpNoDuplicates(t *testing.T) {
t.Parallel()
cfg := &quick.Config{
MaxCount: 1000,
Rand: rand.New(rand.NewSource(time.Now().Unix())),
Values: func(args []reflect.Value, rand *rand.Rand) {
// generate a random list of nodes. this will be the content of the bucket.
n := rand.Intn(bucketSize-1) + 1
nodes := make([]*node, n)
for i := range nodes {
nodes[i] = nodeAtDistance(enode.ID{}, 200, intIP(200))
}
args[0] = reflect.ValueOf(nodes)
// generate random bump positions.
bumps := make([]int, rand.Intn(100))
for i := range bumps {
bumps[i] = rand.Intn(len(nodes))
}
args[1] = reflect.ValueOf(bumps)
},
}
// waitForRevalidationPing waits until a PING message is sent to a node with the given id.
func waitForRevalidationPing(t *testing.T, transport *pingRecorder, tab *Table, id enode.ID) *enode.Node {
t.Helper()
prop := func(nodes []*node, bumps []int) (ok bool) {
tab, db := newTestTable(newPingRecorder())
defer db.Close()
defer tab.close()
b := &bucket{entries: make([]*node, len(nodes))}
copy(b.entries, nodes)
for i, pos := range bumps {
tab.bumpInBucket(b, b.entries[pos])
if hasDuplicates(b.entries) {
t.Logf("bucket has duplicates after %d/%d bumps:", i+1, len(bumps))
for _, n := range b.entries {
t.Logf(" %p", n)
simclock := tab.cfg.Clock.(*mclock.Simulated)
maxAttempts := tab.len() * 8
for i := 0; i < maxAttempts; i++ {
simclock.Run(tab.cfg.PingInterval)
p := transport.waitPing(2 * time.Second)
if p == nil {
t.Fatal("Table did not send revalidation ping")
}
return false
if id == (enode.ID{}) || p.ID() == id {
return p
}
}
checkIPLimitInvariant(t, tab)
return true
}
if err := quick.Check(prop, cfg); err != nil {
t.Error(err)
}
t.Fatalf("Table did not ping node %v (%d attempts)", id, maxAttempts)
return nil
}
// This checks that the table-wide IP limit is applied correctly.
func TestTable_IPLimit(t *testing.T) {
transport := newPingRecorder()
tab, db := newTestTable(transport)
tab, db := newTestTable(transport, Config{})
defer db.Close()
defer tab.close()
for i := 0; i < tableIPLimit+1; i++ {
n := nodeAtDistance(tab.self().ID(), i, net.IP{172, 0, 1, byte(i)})
tab.addSeenNode(n)
tab.addFoundNode(n)
}
if tab.len() > tableIPLimit {
t.Errorf("too many nodes in table")
@ -159,14 +164,14 @@ func TestTable_IPLimit(t *testing.T) {
// This checks that the per-bucket IP limit is applied correctly.
func TestTable_BucketIPLimit(t *testing.T) {
transport := newPingRecorder()
tab, db := newTestTable(transport)
tab, db := newTestTable(transport, Config{})
defer db.Close()
defer tab.close()
d := 3
for i := 0; i < bucketIPLimit+1; i++ {
n := nodeAtDistance(tab.self().ID(), d, net.IP{172, 0, 1, byte(i)})
tab.addSeenNode(n)
tab.addFoundNode(n)
}
if tab.len() > bucketIPLimit {
t.Errorf("too many nodes in table")
@ -196,7 +201,7 @@ func TestTable_findnodeByID(t *testing.T) {
test := func(test *closeTest) bool {
// for any node table, Target and N
transport := newPingRecorder()
tab, db := newTestTable(transport)
tab, db := newTestTable(transport, Config{})
defer db.Close()
defer tab.close()
fillTable(tab, test.All, true)
@ -271,7 +276,7 @@ func (*closeTest) Generate(rand *rand.Rand, size int) reflect.Value {
}
func TestTable_addVerifiedNode(t *testing.T) {
tab, db := newTestTable(newPingRecorder())
tab, db := newTestTable(newPingRecorder(), Config{})
<-tab.initDone
defer db.Close()
defer tab.close()
@ -279,31 +284,32 @@ func TestTable_addVerifiedNode(t *testing.T) {
// Insert two nodes.
n1 := nodeAtDistance(tab.self().ID(), 256, net.IP{88, 77, 66, 1})
n2 := nodeAtDistance(tab.self().ID(), 256, net.IP{88, 77, 66, 2})
tab.addSeenNode(n1)
tab.addSeenNode(n2)
tab.addFoundNode(n1)
tab.addFoundNode(n2)
bucket := tab.bucket(n1.ID())
// Verify bucket content:
bcontent := []*node{n1, n2}
if !reflect.DeepEqual(tab.bucket(n1.ID()).entries, bcontent) {
t.Fatalf("wrong bucket content: %v", tab.bucket(n1.ID()).entries)
if !reflect.DeepEqual(unwrapNodes(bucket.entries), unwrapNodes(bcontent)) {
t.Fatalf("wrong bucket content: %v", bucket.entries)
}
// Add a changed version of n2.
newrec := n2.Record()
newrec.Set(enr.IP{99, 99, 99, 99})
newn2 := wrapNode(enode.SignNull(newrec, n2.ID()))
tab.addVerifiedNode(newn2)
tab.addInboundNode(newn2)
// Check that bucket is updated correctly.
newBcontent := []*node{newn2, n1}
if !reflect.DeepEqual(tab.bucket(n1.ID()).entries, newBcontent) {
t.Fatalf("wrong bucket content after update: %v", tab.bucket(n1.ID()).entries)
newBcontent := []*node{n1, newn2}
if !reflect.DeepEqual(unwrapNodes(bucket.entries), unwrapNodes(newBcontent)) {
t.Fatalf("wrong bucket content after update: %v", bucket.entries)
}
checkIPLimitInvariant(t, tab)
}
func TestTable_addSeenNode(t *testing.T) {
tab, db := newTestTable(newPingRecorder())
tab, db := newTestTable(newPingRecorder(), Config{})
<-tab.initDone
defer db.Close()
defer tab.close()
@ -311,8 +317,8 @@ func TestTable_addSeenNode(t *testing.T) {
// Insert two nodes.
n1 := nodeAtDistance(tab.self().ID(), 256, net.IP{88, 77, 66, 1})
n2 := nodeAtDistance(tab.self().ID(), 256, net.IP{88, 77, 66, 2})
tab.addSeenNode(n1)
tab.addSeenNode(n2)
tab.addFoundNode(n1)
tab.addFoundNode(n2)
// Verify bucket content:
bcontent := []*node{n1, n2}
@ -324,7 +330,7 @@ func TestTable_addSeenNode(t *testing.T) {
newrec := n2.Record()
newrec.Set(enr.IP{99, 99, 99, 99})
newn2 := wrapNode(enode.SignNull(newrec, n2.ID()))
tab.addSeenNode(newn2)
tab.addFoundNode(newn2)
// Check that bucket content is unchanged.
if !reflect.DeepEqual(tab.bucket(n1.ID()).entries, bcontent) {
@ -337,7 +343,10 @@ func TestTable_addSeenNode(t *testing.T) {
// announces a new sequence number, the new record should be pulled.
func TestTable_revalidateSyncRecord(t *testing.T) {
transport := newPingRecorder()
tab, db := newTestTable(transport)
tab, db := newTestTable(transport, Config{
Clock: new(mclock.Simulated),
Log: testlog.Logger(t, log.LevelTrace),
})
<-tab.initDone
defer db.Close()
defer tab.close()
@ -347,14 +356,18 @@ func TestTable_revalidateSyncRecord(t *testing.T) {
r.Set(enr.IP(net.IP{127, 0, 0, 1}))
id := enode.ID{1}
n1 := wrapNode(enode.SignNull(&r, id))
tab.addSeenNode(n1)
tab.addFoundNode(n1)
// Update the node record.
r.Set(enr.WithEntry("foo", "bar"))
n2 := enode.SignNull(&r, id)
transport.updateRecord(n2)
tab.doRevalidate(make(chan struct{}, 1))
// Wait for revalidation. We wait for the node to be revalidated two times
// in order to synchronize with the update in the able.
waitForRevalidationPing(t, transport, tab, n2.ID())
waitForRevalidationPing(t, transport, tab, n2.ID())
intable := tab.getNode(id)
if !reflect.DeepEqual(intable, n2) {
t.Fatalf("table contains old record with seq %d, want seq %d", intable.Seq(), n2.Seq())

112
p2p/discover/table_util_test.go
View file

@ -26,6 +26,8 @@ import (
"net"
"slices"
"sync"
"sync/atomic"
"time"
"github.com/ethereum/go-ethereum/crypto"
"github.com/ethereum/go-ethereum/p2p/enode"
@ -40,8 +42,7 @@ func init() {
nullNode = enode.SignNull(&r, enode.ID{})
}
func newTestTable(t transport) (*Table, *enode.DB) {
cfg := Config{}
func newTestTable(t transport, cfg Config) (*Table, *enode.DB) {
db, _ := enode.OpenDB("")
tab, _ := newTable(t, db, cfg)
go tab.loop()
@ -98,11 +99,14 @@ func intIP(i int) net.IP {
}
// fillBucket inserts nodes into the given bucket until it is full.
func fillBucket(tab *Table, n *node) (last *node) {
ld := enode.LogDist(tab.self().ID(), n.ID())
b := tab.bucket(n.ID())
func fillBucket(tab *Table, id enode.ID) (last *node) {
ld := enode.LogDist(tab.self().ID(), id)
b := tab.bucket(id)
for len(b.entries) < bucketSize {
b.entries = append(b.entries, nodeAtDistance(tab.self().ID(), ld, intIP(ld)))
node := nodeAtDistance(tab.self().ID(), ld, intIP(ld))
if !tab.addFoundNode(node) {
panic("node not added")
}
}
return b.entries[bucketSize-1]
}
@ -113,15 +117,18 @@ func fillTable(tab *Table, nodes []*node, setLive bool) {
for _, n := range nodes {
if setLive {
n.livenessChecks = 1
n.isValidatedLive = true
}
tab.addSeenNode(n)
tab.addFoundNode(n)
}
}
type pingRecorder struct {
mu sync.Mutex
dead, pinged map[enode.ID]bool
cond *sync.Cond
dead map[enode.ID]bool
records map[enode.ID]*enode.Node
pinged []*enode.Node
n *enode.Node
}
@ -130,12 +137,13 @@ func newPingRecorder() *pingRecorder {
r.Set(enr.IP{0, 0, 0, 0})
n := enode.SignNull(&r, enode.ID{})
return &pingRecorder{
t := &pingRecorder{
dead: make(map[enode.ID]bool),
pinged: make(map[enode.ID]bool),
records: make(map[enode.ID]*enode.Node),
n: n,
}
t.cond = sync.NewCond(&t.mu)
return t
}
// updateRecord updates a node record. Future calls to ping and
@ -151,12 +159,40 @@ func (t *pingRecorder) Self() *enode.Node { return nullNode }
func (t *pingRecorder) lookupSelf() []*enode.Node { return nil }
func (t *pingRecorder) lookupRandom() []*enode.Node { return nil }
func (t *pingRecorder) waitPing(timeout time.Duration) *enode.Node {
t.mu.Lock()
defer t.mu.Unlock()
// Wake up the loop on timeout.
var timedout atomic.Bool
timer := time.AfterFunc(timeout, func() {
timedout.Store(true)
t.cond.Broadcast()
})
defer timer.Stop()
// Wait for a ping.
for {
if timedout.Load() {
return nil
}
if len(t.pinged) > 0 {
n := t.pinged[0]
t.pinged = append(t.pinged[:0], t.pinged[1:]...)
return n
}
t.cond.Wait()
}
}
// ping simulates a ping request.
func (t *pingRecorder) ping(n *enode.Node) (seq uint64, err error) {
t.mu.Lock()
defer t.mu.Unlock()
t.pinged[n.ID()] = true
t.pinged = append(t.pinged, n)
t.cond.Broadcast()
if t.dead[n.ID()] {
return 0, errTimeout
}
@ -256,3 +292,57 @@ func hexEncPubkey(h string) (ret encPubkey) {
copy(ret[:], b)
return ret
}
type nodeEventRecorder struct {
evc chan recordedNodeEvent
}
type recordedNodeEvent struct {
node *node
added bool
}
func newNodeEventRecorder(buffer int) *nodeEventRecorder {
return &nodeEventRecorder{
evc: make(chan recordedNodeEvent, buffer),
}
}
func (set *nodeEventRecorder) nodeAdded(b *bucket, n *node) {
select {
case set.evc <- recordedNodeEvent{n, true}:
default:
panic("no space in event buffer")
}
}
func (set *nodeEventRecorder) nodeRemoved(b *bucket, n *node) {
select {
case set.evc <- recordedNodeEvent{n, false}:
default:
panic("no space in event buffer")
}
}
func (set *nodeEventRecorder) waitNodePresent(id enode.ID, timeout time.Duration) bool {
return set.waitNodeEvent(id, timeout, true)
}
func (set *nodeEventRecorder) waitNodeAbsent(id enode.ID, timeout time.Duration) bool {
return set.waitNodeEvent(id, timeout, false)
}
func (set *nodeEventRecorder) waitNodeEvent(id enode.ID, timeout time.Duration, added bool) bool {
timer := time.NewTimer(timeout)
defer timer.Stop()
for {
select {
case ev := <-set.evc:
if ev.node.ID() == id && ev.added == added {
return true
}
case <-timer.C:
return false
}
}
}

10
p2p/discover/v4_udp.go
View file

@ -142,7 +142,7 @@ func ListenV4(c UDPConn, ln *enode.LocalNode, cfg Config) (*UDPv4, error) {
log: cfg.Log,
}
tab, err := newMeteredTable(t, ln.Database(), cfg)
tab, err := newTable(t, ln.Database(), cfg)
if err != nil {
return nil, err
}
@ -375,6 +375,10 @@ func (t *UDPv4) RequestENR(n *enode.Node) (*enode.Node, error) {
return respN, nil
}
func (t *UDPv4) TableBuckets() [][]BucketNode {
return t.tab.Nodes()
}
// pending adds a reply matcher to the pending reply queue.
// see the documentation of type replyMatcher for a detailed explanation.
func (t *UDPv4) pending(id enode.ID, ip net.IP, ptype byte, callback replyMatchFunc) *replyMatcher {
@ -669,10 +673,10 @@ func (t *UDPv4) handlePing(h *packetHandlerV4, from *net.UDPAddr, fromID enode.I
n := wrapNode(enode.NewV4(h.senderKey, from.IP, int(req.From.TCP), from.Port))
if time.Since(t.db.LastPongReceived(n.ID(), from.IP)) > bondExpiration {
t.sendPing(fromID, from, func() {
t.tab.addVerifiedNode(n)
t.tab.addInboundNode(n)
})
} else {
t.tab.addVerifiedNode(n)
t.tab.addInboundNode(n)
}
// Update node database and endpoint predictor.

2
p2p/discover/v4_udp_test.go
View file

@ -264,7 +264,7 @@ func TestUDPv4_findnode(t *testing.T) {
n := wrapNode(enode.NewV4(&key.PublicKey, ip, 0, 2000))
// Ensure half of table content isn't verified live yet.
if i > numCandidates/2 {
n.livenessChecks = 1
n.isValidatedLive = true
live[n.ID()] = true
}
nodes.push(n, numCandidates)

4
p2p/discover/v5_udp.go
View file

@ -175,7 +175,7 @@ func newUDPv5(conn UDPConn, ln *enode.LocalNode, cfg Config) (*UDPv5, error) {
cancelCloseCtx: cancelCloseCtx,
}
t.talk = newTalkSystem(t)
tab, err := newMeteredTable(t, t.db, cfg)
tab, err := newTable(t, t.db, cfg)
if err != nil {
return nil, err
}
@ -699,7 +699,7 @@ func (t *UDPv5) handlePacket(rawpacket []byte, fromAddr *net.UDPAddr) error {
}
if fromNode != nil {
// Handshake succeeded, add to table.
t.tab.addSeenNode(wrapNode(fromNode))
t.tab.addInboundNode(wrapNode(fromNode))
}
if packet.Kind() != v5wire.WhoareyouPacket {
// WHOAREYOU logged separately to report errors.

2
p2p/discover/v5_udp_test.go
View file

@ -141,7 +141,7 @@ func TestUDPv5_unknownPacket(t *testing.T) {
// Make node known.
n := test.getNode(test.remotekey, test.remoteaddr).Node()
test.table.addSeenNode(wrapNode(n))
test.table.addFoundNode(wrapNode(n))
test.packetIn(&v5wire.Unknown{Nonce: nonce})
test.waitPacketOut(func(p *v5wire.Whoareyou, addr *net.UDPAddr, _ v5wire.Nonce) {

2
p2p/enode/idscheme.go
View file

@ -157,5 +157,5 @@ func SignNull(r *enr.Record, id ID) *Node {
if err := r.SetSig(NullID{}, []byte{}); err != nil {
panic(err)
}
return &Node{r: *r, id: id}
return newNodeWithID(r, id)
}

137
p2p/enode/node.go
View file

@ -24,6 +24,7 @@ import (
"fmt"
"math/bits"
"net"
"net/netip"
"strings"
"github.com/ethereum/go-ethereum/p2p/enr"
@ -36,6 +37,10 @@ var errMissingPrefix = errors.New("missing 'enr:' prefix for base64-encoded reco
type Node struct {
r enr.Record
id ID
// endpoint information
ip netip.Addr
udp uint16
tcp uint16
}
// New wraps a node record. The record must be valid according to the given
@ -44,11 +49,76 @@ func New(validSchemes enr.IdentityScheme, r *enr.Record) (*Node, error) {
if err := r.VerifySignature(validSchemes); err != nil {
return nil, err
}
node := &Node{r: *r}
if n := copy(node.id[:], validSchemes.NodeAddr(&node.r)); n != len(ID{}) {
return nil, fmt.Errorf("invalid node ID length %d, need %d", n, len(ID{}))
var id ID
if n := copy(id[:], validSchemes.NodeAddr(r)); n != len(id) {
return nil, fmt.Errorf("invalid node ID length %d, need %d", n, len(id))
}
return newNodeWithID(r, id), nil
}
func newNodeWithID(r *enr.Record, id ID) *Node {
n := &Node{r: *r, id: id}
// Set the preferred endpoint.
// Here we decide between IPv4 and IPv6, choosing the 'most global' address.
var ip4 netip.Addr
var ip6 netip.Addr
n.Load((*enr.IPv4Addr)(&ip4))
n.Load((*enr.IPv6Addr)(&ip6))
valid4 := validIP(ip4)
valid6 := validIP(ip6)
switch {
case valid4 && valid6:
if localityScore(ip4) >= localityScore(ip6) {
n.setIP4(ip4)
} else {
n.setIP6(ip6)
}
case valid4:
n.setIP4(ip4)
case valid6:
n.setIP6(ip6)
}
return n
}
// validIP reports whether 'ip' is a valid node endpoint IP address.
func validIP(ip netip.Addr) bool {
return ip.IsValid() && !ip.IsMulticast()
}
func localityScore(ip netip.Addr) int {
switch {
case ip.IsUnspecified():
return 0
case ip.IsLoopback():
return 1
case ip.IsLinkLocalUnicast():
return 2
case ip.IsPrivate():
return 3
default:
return 4
}
}
func (n *Node) setIP4(ip netip.Addr) {
n.ip = ip
n.Load((*enr.UDP)(&n.udp))
n.Load((*enr.TCP)(&n.tcp))
}
func (n *Node) setIP6(ip netip.Addr) {
if ip.Is4In6() {
n.setIP4(ip)
return
}
n.ip = ip
if err := n.Load((*enr.UDP6)(&n.udp)); err != nil {
n.Load((*enr.UDP)(&n.udp))
}
if err := n.Load((*enr.TCP6)(&n.tcp)); err != nil {
n.Load((*enr.TCP)(&n.tcp))
}
return node, nil
}
// MustParse parses a node record or enode:// URL. It panics if the input is invalid.
@ -89,43 +159,45 @@ func (n *Node) Seq() uint64 {
return n.r.Seq()
}
// Incomplete returns true for nodes with no IP address.
func (n *Node) Incomplete() bool {
return n.IP() == nil
}
// Load retrieves an entry from the underlying record.
func (n *Node) Load(k enr.Entry) error {
return n.r.Load(k)
}
// IP returns the IP address of the node. This prefers IPv4 addresses.
// IP returns the IP address of the node.
func (n *Node) IP() net.IP {
var (
ip4 enr.IPv4
ip6 enr.IPv6
)
if n.Load(&ip4) == nil {
return net.IP(ip4)
}
if n.Load(&ip6) == nil {
return net.IP(ip6)
}
return nil
return net.IP(n.ip.AsSlice())
}
// IPAddr returns the IP address of the node.
func (n *Node) IPAddr() netip.Addr {
return n.ip
}
// UDP returns the UDP port of the node.
func (n *Node) UDP() int {
var port enr.UDP
n.Load(&port)
return int(port)
return int(n.udp)
}
// TCP returns the TCP port of the node.
func (n *Node) TCP() int {
var port enr.TCP
n.Load(&port)
return int(port)
return int(n.tcp)
}
// UDPEndpoint returns the announced UDP endpoint.
func (n *Node) UDPEndpoint() (netip.AddrPort, bool) {
if !n.ip.IsValid() || n.ip.IsUnspecified() || n.udp == 0 {
return netip.AddrPort{}, false
}
return netip.AddrPortFrom(n.ip, n.udp), true
}
// TCPEndpoint returns the announced TCP endpoint.
func (n *Node) TCPEndpoint() (netip.AddrPort, bool) {
if !n.ip.IsValid() || n.ip.IsUnspecified() || n.tcp == 0 {
return netip.AddrPort{}, false
}
return netip.AddrPortFrom(n.ip, n.tcp), true
}
// Pubkey returns the secp256k1 public key of the node, if present.
@ -147,16 +219,15 @@ func (n *Node) Record() *enr.Record {
// ValidateComplete checks whether n has a valid IP and UDP port.
// Deprecated: don't use this method.
func (n *Node) ValidateComplete() error {
if n.Incomplete() {
if !n.ip.IsValid() {
return errors.New("missing IP address")
}
if n.UDP() == 0 {
return errors.New("missing UDP port")
}
ip := n.IP()
if ip.IsMulticast() || ip.IsUnspecified() {
if n.ip.IsMulticast() || n.ip.IsUnspecified() {
return errors.New("invalid IP (multicast/unspecified)")
}
if n.udp == 0 {
return errors.New("missing UDP port")
}
// Validate the node key (on curve, etc.).
var key Secp256k1
return n.Load(&key)

162
p2p/enode/node_test.go
View file

@ -21,6 +21,7 @@ import (
"encoding/hex"
"fmt"
"math/big"
"net/netip"
"testing"
"testing/quick"
@ -64,6 +65,167 @@ func TestPythonInterop(t *testing.T) {
}
}
func TestNodeEndpoints(t *testing.T) {
id := HexID("00000000000000806ad9b61fa5ae014307ebdc964253adcd9f2c0a392aa11abc")
type endpointTest struct {
name string
node *Node
wantIP netip.Addr
wantUDP int
wantTCP int
}
tests := []endpointTest{
{
name: "no-addr",
node: func() *Node {
var r enr.Record
return SignNull(&r, id)
}(),
},
{
name: "udp-only",
node: func() *Node {
var r enr.Record
r.Set(enr.UDP(9000))
return SignNull(&r, id)
}(),
},
{
name: "tcp-only",
node: func() *Node {
var r enr.Record
r.Set(enr.TCP(9000))
return SignNull(&r, id)
}(),
},
{
name: "ipv4-only-loopback",
node: func() *Node {
var r enr.Record
r.Set(enr.IPv4Addr(netip.MustParseAddr("127.0.0.1")))
return SignNull(&r, id)
}(),
wantIP: netip.MustParseAddr("127.0.0.1"),
},
{
name: "ipv4-only-unspecified",
node: func() *Node {
var r enr.Record
r.Set(enr.IPv4Addr(netip.MustParseAddr("0.0.0.0")))
return SignNull(&r, id)
}(),
wantIP: netip.MustParseAddr("0.0.0.0"),
},
{
name: "ipv4-only",
node: func() *Node {
var r enr.Record
r.Set(enr.IPv4Addr(netip.MustParseAddr("99.22.33.1")))
return SignNull(&r, id)
}(),
wantIP: netip.MustParseAddr("99.22.33.1"),
},
{
name: "ipv6-only",
node: func() *Node {
var r enr.Record
r.Set(enr.IPv6Addr(netip.MustParseAddr("2001::ff00:0042:8329")))
return SignNull(&r, id)
}(),
wantIP: netip.MustParseAddr("2001::ff00:0042:8329"),
},
{
name: "ipv4-loopback-and-ipv6-global",
node: func() *Node {
var r enr.Record
r.Set(enr.IPv4Addr(netip.MustParseAddr("127.0.0.1")))
r.Set(enr.UDP(30304))
r.Set(enr.IPv6Addr(netip.MustParseAddr("2001::ff00:0042:8329")))
r.Set(enr.UDP6(30306))
return SignNull(&r, id)
}(),
wantIP: netip.MustParseAddr("2001::ff00:0042:8329"),
wantUDP: 30306,
},
{
name: "ipv4-unspecified-and-ipv6-loopback",
node: func() *Node {
var r enr.Record
r.Set(enr.IPv4Addr(netip.MustParseAddr("0.0.0.0")))
r.Set(enr.IPv6Addr(netip.MustParseAddr("::1")))
return SignNull(&r, id)
}(),
wantIP: netip.MustParseAddr("::1"),
},
{
name: "ipv4-private-and-ipv6-global",
node: func() *Node {
var r enr.Record
r.Set(enr.IPv4Addr(netip.MustParseAddr("192.168.2.2")))
r.Set(enr.UDP(30304))
r.Set(enr.IPv6Addr(netip.MustParseAddr("2001::ff00:0042:8329")))
r.Set(enr.UDP6(30306))
return SignNull(&r, id)
}(),
wantIP: netip.MustParseAddr("2001::ff00:0042:8329"),
wantUDP: 30306,
},
{
name: "ipv4-local-and-ipv6-global",
node: func() *Node {
var r enr.Record
r.Set(enr.IPv4Addr(netip.MustParseAddr("169.254.2.6")))
r.Set(enr.UDP(30304))
r.Set(enr.IPv6Addr(netip.MustParseAddr("2001::ff00:0042:8329")))
r.Set(enr.UDP6(30306))
return SignNull(&r, id)
}(),
wantIP: netip.MustParseAddr("2001::ff00:0042:8329"),
wantUDP: 30306,
},
{
name: "ipv4-private-and-ipv6-private",
node: func() *Node {
var r enr.Record
r.Set(enr.IPv4Addr(netip.MustParseAddr("192.168.2.2")))
r.Set(enr.UDP(30304))
r.Set(enr.IPv6Addr(netip.MustParseAddr("fd00::abcd:1")))
r.Set(enr.UDP6(30306))
return SignNull(&r, id)
}(),
wantIP: netip.MustParseAddr("192.168.2.2"),
wantUDP: 30304,
},
{
name: "ipv4-private-and-ipv6-link-local",
node: func() *Node {
var r enr.Record
r.Set(enr.IPv4Addr(netip.MustParseAddr("192.168.2.2")))
r.Set(enr.UDP(30304))
r.Set(enr.IPv6Addr(netip.MustParseAddr("fe80::1")))
r.Set(enr.UDP6(30306))
return SignNull(&r, id)
}(),
wantIP: netip.MustParseAddr("192.168.2.2"),
wantUDP: 30304,
},
}
for _, test := range tests {
t.Run(test.name, func(t *testing.T) {
if test.wantIP != test.node.IPAddr() {
t.Errorf("node has wrong IP %v, want %v", test.node.IPAddr(), test.wantIP)
}
if test.wantUDP != test.node.UDP() {
t.Errorf("node has wrong UDP port %d, want %d", test.node.UDP(), test.wantUDP)
}
if test.wantTCP != test.node.TCP() {
t.Errorf("node has wrong TCP port %d, want %d", test.node.TCP(), test.wantTCP)
}
})
}
}
func TestHexID(t *testing.T) {
ref := ID{0, 0, 0, 0, 0, 0, 0, 128, 106, 217, 182, 31, 165, 174, 1, 67, 7, 235, 220, 150, 66, 83, 173, 205, 159, 44, 10, 57, 42, 161, 26, 188}
id1 := HexID("0x00000000000000806ad9b61fa5ae014307ebdc964253adcd9f2c0a392aa11abc")

12
p2p/enode/nodedb.go
View file

@ -26,6 +26,7 @@ import (
"sync"
"time"
"github.com/ethereum/go-ethereum/p2p/enr"
"github.com/ethereum/go-ethereum/rlp"
"github.com/syndtr/goleveldb/leveldb"
"github.com/syndtr/goleveldb/leveldb/errors"
@ -242,13 +243,14 @@ func (db *DB) Node(id ID) *Node {
}
func mustDecodeNode(id, data []byte) *Node {
node := new(Node)
if err := rlp.DecodeBytes(data, &node.r); err != nil {
var r enr.Record
if err := rlp.DecodeBytes(data, &r); err != nil {
panic(fmt.Errorf("p2p/enode: can't decode node %x in DB: %v", id, err))
}
// Restore node id cache.
copy(node.id[:], id)
return node
if len(id) != len(ID{}) {
panic(fmt.Errorf("invalid id length %d", len(id)))
}
return newNodeWithID(&r, ID(id))
}
// UpdateNode inserts - potentially overwriting - a node into the peer database.

2
p2p/enode/urlv4.go
View file

@ -181,7 +181,7 @@ func (n *Node) URLv4() string {
nodeid = fmt.Sprintf("%s.%x", scheme, n.id[:])
}
u := url.URL{Scheme: "enode"}
if n.Incomplete() {
if !n.ip.IsValid() {
u.Host = nodeid
} else {
addr := net.TCPAddr{IP: n.IP(), Port: n.TCP()}

55
p2p/enr/entries.go
View file

@ -21,6 +21,7 @@ import (
"fmt"
"io"
"net"
"net/netip"
"github.com/ethereum/go-ethereum/rlp"
)
@ -167,6 +168,60 @@ func (v *IPv6) DecodeRLP(s *rlp.Stream) error {
return nil
}
// IPv4Addr is the "ip" key, which holds the IP address of the node.
type IPv4Addr netip.Addr
func (v IPv4Addr) ENRKey() string { return "ip" }
// EncodeRLP implements rlp.Encoder.
func (v IPv4Addr) EncodeRLP(w io.Writer) error {
addr := netip.Addr(v)
if !addr.Is4() {
return fmt.Errorf("address is not IPv4")
}
enc := rlp.NewEncoderBuffer(w)
bytes := addr.As4()
enc.WriteBytes(bytes[:])
return enc.Flush()
}
// DecodeRLP implements rlp.Decoder.
func (v *IPv4Addr) DecodeRLP(s *rlp.Stream) error {
var bytes [4]byte
if err := s.ReadBytes(bytes[:]); err != nil {
return err
}
*v = IPv4Addr(netip.AddrFrom4(bytes))
return nil
}
// IPv6Addr is the "ip6" key, which holds the IP address of the node.
type IPv6Addr netip.Addr
func (v IPv6Addr) ENRKey() string { return "ip6" }
// EncodeRLP implements rlp.Encoder.
func (v IPv6Addr) EncodeRLP(w io.Writer) error {
addr := netip.Addr(v)
if !addr.Is6() {
return fmt.Errorf("address is not IPv6")
}
enc := rlp.NewEncoderBuffer(w)
bytes := addr.As16()
enc.WriteBytes(bytes[:])
return enc.Flush()
}
// DecodeRLP implements rlp.Decoder.
func (v *IPv6Addr) DecodeRLP(s *rlp.Stream) error {
var bytes [16]byte
if err := s.ReadBytes(bytes[:]); err != nil {
return err
}
*v = IPv6Addr(netip.AddrFrom16(bytes))
return nil
}
// KeyError is an error related to a key.
type KeyError struct {
Key string

36
p2p/server.go
View file

@ -190,8 +190,8 @@ type Server struct {
nodedb *enode.DB
localnode *enode.LocalNode
ntab *discover.UDPv4
DiscV5 *discover.UDPv5
discv4 *discover.UDPv4
discv5 *discover.UDPv5
discmix *enode.FairMix
dialsched *dialScheduler
@ -400,6 +400,16 @@ func (srv *Server) Self() *enode.Node {
return ln.Node()
}
// DiscoveryV4 returns the discovery v4 instance, if configured.
func (srv *Server) DiscoveryV4() *discover.UDPv4 {
return srv.discv4
}
// DiscoveryV4 returns the discovery v5 instance, if configured.
func (srv *Server) DiscoveryV5() *discover.UDPv5 {
return srv.discv5
}
// Stop terminates the server and all active peer connections.
// It blocks until all active connections have been closed.
func (srv *Server) Stop() {
@ -547,13 +557,13 @@ func (srv *Server) setupDiscovery() error {
)
// If both versions of discovery are running, setup a shared
// connection, so v5 can read unhandled messages from v4.
if srv.DiscoveryV4 && srv.DiscoveryV5 {
if srv.Config.DiscoveryV4 && srv.Config.DiscoveryV5 {
unhandled = make(chan discover.ReadPacket, 100)
sconn = &sharedUDPConn{conn, unhandled}
}
// Start discovery services.
if srv.DiscoveryV4 {
if srv.Config.DiscoveryV4 {
cfg := discover.Config{
PrivateKey: srv.PrivateKey,
NetRestrict: srv.NetRestrict,
@ -565,17 +575,17 @@ func (srv *Server) setupDiscovery() error {
if err != nil {
return err
}
srv.ntab = ntab
srv.discv4 = ntab
srv.discmix.AddSource(ntab.RandomNodes())
}
if srv.DiscoveryV5 {
if srv.Config.DiscoveryV5 {
cfg := discover.Config{
PrivateKey: srv.PrivateKey,
NetRestrict: srv.NetRestrict,
Bootnodes: srv.BootstrapNodesV5,
Log: srv.log,
}
srv.DiscV5, err = discover.ListenV5(sconn, srv.localnode, cfg)
srv.discv5, err = discover.ListenV5(sconn, srv.localnode, cfg)
if err != nil {
return err
}
@ -602,8 +612,8 @@ func (srv *Server) setupDialScheduler() {
dialer: srv.Dialer,
clock: srv.clock,
}
if srv.ntab != nil {
config.resolver = srv.ntab
if srv.discv4 != nil {
config.resolver = srv.discv4
}
if config.dialer == nil {
config.dialer = tcpDialer{&net.Dialer{Timeout: defaultDialTimeout}}
@ -799,11 +809,11 @@ running:
srv.log.Trace("P2P networking is spinning down")
// Terminate discovery. If there is a running lookup it will terminate soon.
if srv.ntab != nil {
srv.ntab.Close()
if srv.discv4 != nil {
srv.discv4.Close()
}
if srv.DiscV5 != nil {
srv.DiscV5.Close()
if srv.discv5 != nil {
srv.discv5.Close()
}
// Disconnect all peers.
for _, p := range peers {

4
params/protocol_params.go
View file

@ -187,6 +187,10 @@ var (
// BeaconRootsAddress is the address where historical beacon roots are stored as per EIP-4788
BeaconRootsAddress = common.HexToAddress("0x000F3df6D732807Ef1319fB7B8bB8522d0Beac02")
// BeaconRootsCode is the code where historical beacon roots are stored as per EIP-4788
BeaconRootsCode = common.FromHex("3373fffffffffffffffffffffffffffffffffffffffe14604d57602036146024575f5ffd5b5f35801560495762001fff810690815414603c575f5ffd5b62001fff01545f5260205ff35b5f5ffd5b62001fff42064281555f359062001fff015500")
// SystemAddress is where the system-transaction is sent from as per EIP-4788
SystemAddress = common.HexToAddress("0xfffffffffffffffffffffffffffffffffffffffe")
)

21
tests/init.go
View file

@ -212,7 +212,7 @@ var Forks = map[string]*params.ChainConfig{
LondonBlock: big.NewInt(0),
ArrowGlacierBlock: big.NewInt(0),
},
"ArrowGlacierToMergeAtDiffC0000": {
"ArrowGlacierToParisAtDiffC0000": {
ChainID: big.NewInt(1),
HomesteadBlock: big.NewInt(0),
EIP150Block: big.NewInt(0),
@ -246,6 +246,23 @@ var Forks = map[string]*params.ChainConfig{
ArrowGlacierBlock: big.NewInt(0),
GrayGlacierBlock: big.NewInt(0),
},
"Paris": {
ChainID: big.NewInt(1),
HomesteadBlock: big.NewInt(0),
EIP150Block: big.NewInt(0),
EIP155Block: big.NewInt(0),
EIP158Block: big.NewInt(0),
ByzantiumBlock: big.NewInt(0),
ConstantinopleBlock: big.NewInt(0),
PetersburgBlock: big.NewInt(0),
IstanbulBlock: big.NewInt(0),
MuirGlacierBlock: big.NewInt(0),
BerlinBlock: big.NewInt(0),
LondonBlock: big.NewInt(0),
ArrowGlacierBlock: big.NewInt(0),
MergeNetsplitBlock: big.NewInt(0),
TerminalTotalDifficulty: big.NewInt(0),
},
"Merge": {
ChainID: big.NewInt(1),
HomesteadBlock: big.NewInt(0),
@ -281,7 +298,7 @@ var Forks = map[string]*params.ChainConfig{
TerminalTotalDifficulty: big.NewInt(0),
ShanghaiTime: u64(0),
},
"MergeToShanghaiAtTime15k": {
"ParisToShanghaiAtTime15k": {
ChainID: big.NewInt(1),
HomesteadBlock: big.NewInt(0),
EIP150Block: big.NewInt(0),

8
tests/state_test.go
View file

@ -54,14 +54,6 @@ func initMatcher(st *testMatcher) {
// Uses 1GB RAM per tested fork
st.skipLoad(`^stStaticCall/static_Call1MB`)
// These tests fail as of https://github.com/ethereum/go-ethereum/pull/28666, since we
// no longer delete "leftover storage" when deploying a contract.
st.skipLoad(`^stSStoreTest/InitCollision\.json`)
st.skipLoad(`^stRevertTest/RevertInCreateInInit\.json`)
st.skipLoad(`^stExtCodeHash/dynamicAccountOverwriteEmpty\.json`)
st.skipLoad(`^stCreate2/create2collisionStorage\.json`)
st.skipLoad(`^stCreate2/RevertInCreateInInitCreate2\.json`)
// Broken tests:
// EOF is not part of cancun
st.skipLoad(`^stEOF/`)

2
tests/testdata

@ -1 +1 @@
Subproject commit fa51c5c164f79140730ccb8fe26a46c3d3994338
Subproject commit faf33b471465d3c6cdc3d04fbd690895f78d33f2

40
trie/sync.go
View file

@ -22,6 +22,7 @@ import (
"sync"
"github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/common/hexutil"
"github.com/ethereum/go-ethereum/common/prque"
"github.com/ethereum/go-ethereum/core/rawdb"
"github.com/ethereum/go-ethereum/core/types"
@ -149,15 +150,42 @@ type CodeSyncResult struct {
// nodeOp represents an operation upon the trie node. It can either represent a
// deletion to the specific node or a node write for persisting retrieved node.
type nodeOp struct {
del bool // flag if op stands for a delete operation
owner common.Hash // identifier of the trie (empty for account trie)
path []byte // path from the root to the specified node.
blob []byte // the content of the node (nil for deletion)
hash common.Hash // hash of the node content (empty for node deletion)
}
// isDelete indicates if the operation is a database deletion.
func (op *nodeOp) isDelete() bool {
return len(op.blob) == 0
// valid checks whether the node operation is valid.
func (op *nodeOp) valid() bool {
if op.del && len(op.blob) != 0 {
return false
}
if !op.del && len(op.blob) == 0 {
return false
}
return true
}
// string returns the node operation in string representation.
func (op *nodeOp) string() string {
var node string
if op.owner == (common.Hash{}) {
node = fmt.Sprintf("node: (%v)", op.path)
} else {
node = fmt.Sprintf("node: (%x-%v)", op.owner, op.path)
}
var blobHex string
if len(op.blob) == 0 {
blobHex = "nil"
} else {
blobHex = hexutil.Encode(op.blob)
}
if op.del {
return fmt.Sprintf("del %s %s %s", node, blobHex, op.hash.Hex())
}
return fmt.Sprintf("write %s %s %s", node, blobHex, op.hash.Hex())
}
// syncMemBatch is an in-memory buffer of successfully downloaded but not yet
@ -220,6 +248,7 @@ func (batch *syncMemBatch) delNode(owner common.Hash, path []byte) {
batch.size += common.HashLength + uint64(len(path))
}
batch.nodes = append(batch.nodes, nodeOp{
del: true,
owner: owner,
path: path,
})
@ -428,7 +457,10 @@ func (s *Sync) Commit(dbw ethdb.Batch) error {
storage int
)
for _, op := range s.membatch.nodes {
if op.isDelete() {
if !op.valid() {
return fmt.Errorf("invalid op, %s", op.string())
}
if op.del {
// node deletion is only supported in path mode.
if op.owner == (common.Hash{}) {
rawdb.DeleteAccountTrieNode(dbw, op.path)