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

p2p/discover: new revalidation

Browse source
This commit is contained in:
Felix Lange 2024-04-18 00:02:49 +02:00
parent f8c744894c
commit df8e793e1f
9 changed files with 628 additions and 460 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

35
p2p/discover/common.go
View file

@ -18,7 +18,11 @@ package discover
import (
"crypto/ecdsa"
crand "crypto/rand"
"encoding/binary"
"math/rand"
"net"
"sync/atomic"
"time"
"github.com/ethereum/go-ethereum/common/mclock"
@ -92,3 +96,34 @@ type ReadPacket struct {
Data []byte
Addr *net.UDPAddr
}
type randomSource interface {
Intn(int) int
Int63n(int64) int64
}
// reseedingRandom is a random number generator that tracks when it was last re-seeded.
type reseedingRandom struct {
cur atomic.Pointer[rand.Rand]
lastSeed mclock.AbsTime
}
func (r *reseedingRandom) nextReseedTime() mclock.AbsTime {
return r.lastSeed.Add(10 * time.Minute)
}
func (r *reseedingRandom) seed(now mclock.AbsTime) {
var b [8]byte
crand.Read(b[:])
seed := binary.BigEndian.Uint64(b[:])
r.cur.Store(rand.New(rand.NewSource(int64(seed))))
r.lastSeed = now
}
func (r *reseedingRandom) Intn(n int) int {
return r.cur.Load().Intn(n)
}
func (r *reseedingRandom) Int63n(n int64) int64 {
return r.cur.Load().Int63n(n)
}

4
p2p/discover/lookup.go
View file

@ -152,9 +152,9 @@ func (it *lookup) query(n *node, reply chan<- []*node) {
// 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 {
if fails >= maxFindnodeFailures {
dropped = true
it.tab.delete(n)
it.tab.trackFindFailure(n)
}
it.tab.log.Trace("FINDNODE failed", "id", n.ID(), "failcount", fails, "dropped", dropped, "err", err)
} else if fails > 0 {

11
p2p/discover/node.go
View file

@ -38,9 +38,10 @@ type BucketNode struct {
// 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
livenessChecks uint // how often liveness was checked
*enode.Node
addedAt time.Time // time when the node was added to the table
livenessChecks uint // how often liveness was checked
isValidatedLive bool
}
type encPubkey [64]byte
@ -71,7 +72,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 {
@ -83,7 +84,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 {

638
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,21 +54,21 @@ 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
seedMinTableTime = 5 * time.Minute
seedCount = 30
seedMaxAge = 5 * 24 * time.Hour
)
// Table is the 'node table', a Kademlia-like index of neighbor nodes. The table keeps
// itself up-to-date by verifying the liveness of neighbors and requesting their node
// records when announcements of a new record version are received.
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
ips netutil.DistinctNetSet
mutex sync.Mutex // protects buckets, bucket content, nursery, rand
buckets [nBuckets]*bucket // index of known nodes by distance
nursery []*node // bootstrap nodes
rand *reseedingRandom // source of randomness, periodically reseeded
ips netutil.DistinctNetSet
revalidation tableRevalidation
db *enode.DB // database of known nodes
net transport
@ -77,10 +76,14 @@ type Table struct {
log log.Logger
// loop channels
refreshReq chan chan struct{}
initDone chan struct{}
closeReq chan struct{}
closed chan struct{}
refreshReq chan chan struct{}
revalidateResp chan revalidationResponse
addNodeCh chan addNodeRequest
addNodeHandled chan struct{}
findFailureCh chan *node
initDone chan struct{}
closeReq chan struct{}
closed chan struct{}
nodeAddedHook func(*bucket, *node)
nodeRemovedHook func(*bucket, *node)
@ -104,22 +107,28 @@ type bucket struct {
index int
}
type addNodeRequest struct {
node *node
isLive bool
}
func newTable(t transport, db *enode.DB, cfg Config) (*Table, error) {
cfg = cfg.withDefaults()
tab := &Table{
net: t,
db: db,
cfg: cfg,
log: cfg.Log,
refreshReq: make(chan chan struct{}),
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
net: t,
db: db,
cfg: cfg,
log: cfg.Log,
refreshReq: make(chan chan struct{}),
revalidateResp: make(chan revalidationResponse),
addNodeCh: make(chan addNodeRequest),
addNodeHandled: make(chan struct{}),
findFailureCh: make(chan *node),
initDone: make(chan struct{}),
closeReq: make(chan struct{}),
closed: make(chan struct{}),
rand: new(reseedingRandom),
ips: netutil.DistinctNetSet{Subnet: tableSubnet, Limit: tableIPLimit},
}
for i := range tab.buckets {
tab.buckets[i] = &bucket{
@ -127,26 +136,23 @@ func newTable(t transport, db *enode.DB, cfg Config) (*Table, error) {
ips: netutil.DistinctNetSet{Subnet: bucketSubnet, Limit: bucketIPLimit},
}
}
tab.seedRand()
tab.rand.seed(cfg.Clock.Now())
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
func (tab *Table) trackFindFailure(n *node) {
select {
case tab.findFailureCh <- n:
case <-tab.closed:
}
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.
@ -172,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()
@ -240,52 +237,157 @@ 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()
defer tab.mutex.Unlock()
for _, n := range tab.bucketAtDistance(int(dist)).entries {
if n.isValidatedLive {
result = append(result, n.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
}
// addSeenNode 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) addSeenNode(n *node) {
req := addNodeRequest{node: n, isLive: false}
select {
case tab.addNodeCh <- req:
<-tab.addNodeHandled
case <-tab.closed:
}
}
// addVerifiedNode adds a node whose existence has been verified recently. 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) {
req := addNodeRequest{node: n, isLive: true}
select {
case tab.addNodeCh <- req:
<-tab.addNodeHandled
case <-tab.closed:
}
}
// 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
refresh = time.NewTimer(tab.nextRefreshTime())
refreshDone = make(chan struct{}) // where doRefresh reports completion
waiting = []chan struct{}{tab.initDone} // holds waiting callers while doRefresh runs
revalTimer = mclock.NewAlarm(tab.cfg.Clock)
reseedRandTimer = mclock.NewAlarm(tab.cfg.Clock)
)
defer refresh.Stop()
defer revalidate.Stop()
defer copyNodes.Stop()
defer revalTimer.Stop()
defer reseedRandTimer.Stop()
// Start initial refresh.
go tab.doRefresh(refreshDone)
loop:
for {
reseedRandTimer.Schedule(tab.rand.nextReseedTime())
revalTimer.Schedule(tab.revalidation.nextTime())
select {
case <-reseedRandTimer.C():
tab.rand.seed(tab.cfg.Clock.Now())
case <-revalTimer.C():
tab.revalidation.run(tab, mclock.Now())
case r := <-tab.revalidateResp:
tab.revalidation.handleResponse(tab, r)
case addreq := <-tab.addNodeCh:
tab.handleAddNode(addreq)
tab.addNodeHandled <- struct{}{}
case <-tab.findFailureCh:
// TODO: handle failure by potentially dropping node
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
}
@ -297,9 +399,6 @@ loop:
for _, ch := range waiting {
close(ch)
}
if revalidateDone != nil {
<-revalidateDone
}
close(tab.closed)
}
@ -340,165 +439,11 @@ func (tab *Table) loadSeedNodes() {
}
}
// 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)
@ -512,95 +457,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.
@ -628,11 +484,43 @@ func (tab *Table) removeIP(b *bucket, ip net.IP) {
b.ips.Remove(ip)
}
func (tab *Table) handleAddNode(req addNodeRequest) {
if req.node.ID() == tab.self().ID() {
return
}
if !req.isLive && !tab.isInitDone() {
return
}
tab.mutex.Lock()
defer tab.mutex.Unlock()
b := tab.bucket(req.node.ID())
if tab.bumpInBucket(b, req.node.Node) {
// Already in bucket, update record.
return
}
if len(b.entries) >= bucketSize {
// Bucket full, maybe add as replacement.
tab.addReplacement(b, req.node)
return
}
if !tab.addIP(b, req.node.IP()) {
// Can't add: IP limit reached.
return
}
// Add to bucket.
b.entries, _ = pushNode(b.entries, req.node, bucketSize)
b.replacements = deleteNode(b.replacements, req.node)
req.node.addedAt = time.Now()
tab.nodeAdded(b, req.node)
}
// addReplacement adds n to the replacement cache of bucket b.
func (tab *Table) addReplacement(b *bucket, n *node) {
for _, e := range b.replacements {
if e.ID() == n.ID() {
return // already in list
}
if contains(b.replacements, n.ID()) {
// TODO: update ENR
return
}
if !tab.addIP(b, n.IP()) {
return
@ -644,60 +532,76 @@ 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) {
tab.revalidation.nodeAdded(tab, n)
if tab.nodeAddedHook != nil {
tab.nodeAddedHook(b, n)
}
// Still the last entry.
if len(b.replacements) == 0 {
tab.deleteInBucket(b, last)
return nil
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()) {
// 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()) {
// 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
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
}
b.entries = deleteNode(b.entries, n)
tab.removeIP(b, n.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, don't replace it.
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(), "checks", n.livenessChecks)
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(), "checks", n.livenessChecks, "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, newRecord.IP()) {
// It doesn't, put the previous one back.
tab.addIP(b, b.entries[i].IP())
return false
}
}
b.entries[i].Node = newRecord
return true
}
func contains(ns []*node, id enode.ID) bool {

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

@ -0,0 +1,204 @@
// 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"
"slices"
"time"
"github.com/ethereum/go-ethereum/common/mclock"
"github.com/ethereum/go-ethereum/p2p/enode"
)
const never = ^mclock.AbsTime(0)
type tableRevalidation struct {
newNodes revalidationQueue
nodes revalidationQueue
activeReq map[enode.ID]struct{}
}
type revalidationResponse struct {
n *node
didRespond bool
isNewNode bool
newRecord *enode.Node
}
func (tr *tableRevalidation) init(cfg *Config) {
tr.activeReq = make(map[enode.ID]struct{})
tr.newNodes.nextTime = never
tr.newNodes.interval = cfg.PingInterval
tr.nodes.nextTime = never
tr.nodes.interval = cfg.PingInterval
}
// nodeAdded is called when the table receives a new node.
func (tr *tableRevalidation) nodeAdded(tab *Table, n *node) {
tr.newNodes.push(n, tab.rand)
}
// nodeRemoved is called when a node was removed from the table.
func (tr *tableRevalidation) nodeRemoved(n *node) {
wasnew := tr.newNodes.remove(n)
if !wasnew {
tr.nodes.remove(n)
}
}
// nextTime returns the next time run() should be invoked.
// The Table main loop uses this to schedule a timer.
func (tr *tableRevalidation) nextTime() mclock.AbsTime {
return min(tr.newNodes.nextTime, tr.nodes.nextTime)
}
// run performs node revalidation.
func (tr *tableRevalidation) run(tab *Table, now mclock.AbsTime) {
if n := tr.newNodes.get(now, tab.rand, tr.activeReq); n != nil {
tr.startRequest(tab, n, true)
tr.newNodes.schedule(tab.rand)
}
if n := tr.nodes.get(now, tab.rand, tr.activeReq); n != nil {
tr.startRequest(tab, n, false)
tr.nodes.schedule(tab.rand)
}
}
// startRequest spawns a revalidation request for node n.
func (tr *tableRevalidation) startRequest(tab *Table, n *node, newNode bool) {
if _, ok := tr.activeReq[n.ID()]; ok {
panic("duplicate startRequest")
}
tr.activeReq[n.ID()] = struct{}{}
resp := revalidationResponse{n: n, isNewNode: newNode}
go func() {
// Ping the selected node and wait for a pong response.
remoteSeq, err := tab.net.ping(unwrapNode(n))
resp.didRespond = err == nil
// Also fetch record if the node replied and returned a higher sequence number.
if remoteSeq > n.Seq() {
newrec, err := tab.net.RequestENR(unwrapNode(n))
if err != nil {
tab.log.Debug("ENR request failed", "id", n.ID(), "addr", n.addr(), "err", err)
} else {
resp.newRecord = newrec
}
}
select {
case tab.revalidateResp <- resp:
case <-tab.closed:
}
}()
}
// handleResponse processes the result of a revalidation request.
func (tr *tableRevalidation) handleResponse(tab *Table, resp revalidationResponse) {
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 || resp.isNewNode {
tab.deleteInBucket(b, n.ID())
}
return
}
// The node responded.
n.livenessChecks++
n.isValidatedLive = true
tab.log.Debug("Revalidated node", "b", b.index, "id", n.ID(), "checks", n.livenessChecks)
if resp.newRecord != nil {
updated := tab.bumpInBucket(b, resp.newRecord)
if updated {
// If the node changed its advertised endpoint, the updated ENR is not served
// until it has been revalidated.
n.isValidatedLive = false
}
}
// Move node over to main queue after first validation.
if resp.isNewNode {
tr.newNodes.remove(n)
tr.nodes.push(n, tab.rand)
}
// Store potential seeds in database.
if n.isValidatedLive && n.livenessChecks > 5 {
tab.db.UpdateNode(resp.n.Node)
}
}
// revalidationQueue holds a list nodes and the next revalidation time.
type revalidationQueue struct {
nodes []*node
nextTime mclock.AbsTime
interval time.Duration
}
// get returns a random node from the queue. Nodes in the 'exclude' map are not returned.
func (rq *revalidationQueue) get(now mclock.AbsTime, rand randomSource, exclude map[enode.ID]struct{}) *node {
if now < rq.nextTime || len(rq.nodes) == 0 {
return nil
}
for i := 0; i < len(rq.nodes)*3; i++ {
n := rq.nodes[rand.Intn(len(rq.nodes))]
_, excluded := exclude[n.ID()]
if !excluded {
return n
}
}
return nil
}
func (rq *revalidationQueue) push(n *node, rand randomSource) {
rq.nodes = append(rq.nodes, n)
if rq.nextTime == never {
rq.schedule(rand)
}
}
func (rq *revalidationQueue) schedule(rand randomSource) {
rq.nextTime = mclock.AbsTime(rand.Int63n(int64(rq.interval)))
}
func (rq *revalidationQueue) remove(n *node) bool {
i := slices.Index(rq.nodes, n)
if i == -1 {
return false
}
rq.nodes = slices.Delete(rq.nodes, i, i+1)
if len(rq.nodes) == 0 {
rq.nextTime = never
}
return true
}
func printIDs(list []*node) {
for i, n := range list {
fmt.Println(" - ", i, n.ID())
}
}

132
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,29 +51,34 @@ 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)
// 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
// Add the sender as if it just pinged us. 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
tab.addSeenNode(replacementNode)
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")
// Wait until the last node was pinged.
waitForRevalidationPing(t, transport, tab, last.ID())
// If a replacement is expected, we also need to wait until the replacement node was pinged.
if !lastInBucketIsResponding {
waitForRevalidationPing(t, transport, tab, replacementNode.ID())
}
tab.mutex.Lock()
@ -80,69 +87,41 @@ func testPingReplace(t *testing.T, newNodeIsResponding, lastInBucketIsResponding
if !lastInBucketIsResponding && !newNodeIsResponding {
wantSize--
}
if l := len(tab.bucket(pingSender.ID()).entries); l != wantSize {
bucket := tab.bucket(replacementNode.ID())
if l := len(bucket.entries); l != wantSize {
t.Errorf("wrong bucket size after bond: 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("last entry 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("new entry 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)
},
}
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)
}
return false
}
// 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 {
simclock := tab.cfg.Clock.(*mclock.Simulated)
maxAttempts := tab.len() * 5
for i := 0; i < maxAttempts; i++ {
simclock.Run(tab.cfg.PingInterval)
p := transport.waitPing(500 * time.Millisecond)
if p == nil {
t.Fatal("Table did not send any revalidation ping")
}
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()
@ -159,7 +138,7 @@ 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()
@ -196,7 +175,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 +250,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()
@ -281,11 +260,12 @@ func TestTable_addVerifiedNode(t *testing.T) {
n2 := nodeAtDistance(tab.self().ID(), 256, net.IP{88, 77, 66, 2})
tab.addSeenNode(n1)
tab.addSeenNode(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.
@ -295,15 +275,15 @@ func TestTable_addVerifiedNode(t *testing.T) {
tab.addVerifiedNode(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()
@ -337,7 +317,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()
@ -354,7 +337,8 @@ func TestTable_revalidateSyncRecord(t *testing.T) {
n2 := enode.SignNull(&r, id)
transport.updateRecord(n2)
tab.doRevalidate(make(chan struct{}, 1))
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())

60
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()
@ -102,7 +103,9 @@ func fillBucket(tab *Table, n *node) (last *node) {
ld := enode.LogDist(tab.self().ID(), n.ID())
b := tab.bucket(n.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))
b.entries = append(b.entries, node)
tab.nodeAdded(b, node)
}
return b.entries[bucketSize-1]
}
@ -119,10 +122,12 @@ func fillTable(tab *Table, nodes []*node, setLive bool) {
}
type pingRecorder struct {
mu sync.Mutex
dead, pinged map[enode.ID]bool
records map[enode.ID]*enode.Node
n *enode.Node
mu sync.Mutex
cond *sync.Cond
dead map[enode.ID]bool
records map[enode.ID]*enode.Node
pinged []*enode.Node
n *enode.Node
}
func newPingRecorder() *pingRecorder {
@ -130,12 +135,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 +157,46 @@ 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) wasPinged(id enode.ID) bool {
t.mu.Lock()
defer t.mu.Unlock()
return slices.ContainsFunc(t.pinged, func(n *enode.Node) bool { return n.ID() == id })
}
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
}

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

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