p2p/discover: improved node revalidation (ethereum#29572)

Node discovery periodically revalidates the nodes in its table by sending PING, checking
if they are still alive. I recently noticed some issues with the implementation of this
process, which can cause strange results such as nodes dropping unexpectedly, certain
nodes not getting revalidated often enough, and bad results being returned to incoming
FINDNODE queries.

In this change, the revalidation process is improved with the following logic:

- We maintain two 'revalidation lists' containing the table nodes, named 'fast' and 'slow'.
- The process chooses random nodes from each list on a randomized interval, the interval being
  faster for the 'fast' list, and performs revalidation for the chosen node.
- Whenever a node is newly inserted into the table, it goes into the 'fast' list.
  Once validation passes, it transfers to the 'slow' list. If a request fails, or the
  node changes endpoint, it transfers back into 'fast'.
- livenessChecks is incremented by one for successful checks. Unlike the old implementation,
  we will not drop the node on the first failing check. We instead quickly decay the
  livenessChecks give it another chance.
- Order of nodes in bucket doesn't matter anymore.

I am also adding a debug API endpoint to dump the node table content.

Co-authored-by: Martin HS <[email protected]>

Author: 2 people

les/client.go

@@ -218,7 +218,7 @@ func (s *LightEthereum) VfluxRequest(n *enode.Node, reqs vflux.Requests) vflux.R
 		return nil
 	}
 	reqsEnc, _ := rlp.EncodeToBytes(&reqs)
-	repliesEnc, _ := s.p2pServer.DiscV5.TalkRequest(s.serverPool.DialNode(n), "vfx", reqsEnc)
+	repliesEnc, _ := s.p2pServer.DiscoveryV5().TalkRequest(s.serverPool.DialNode(n), "vfx", reqsEnc)
 	var replies vflux.Replies
 	if len(repliesEnc) == 0 || rlp.DecodeBytes(repliesEnc, &replies) != nil {
 		return nil
@@ -234,7 +234,7 @@ func (s *LightEthereum) vfxVersion(n *enode.Node) uint {
 		if !s.udpEnabled {
 			return 0
 		}
-		if n, err = s.p2pServer.DiscV5.RequestENR(n); n != nil && err == nil && n.Seq() != 0 {
+		if n, err = s.p2pServer.DiscoveryV5().RequestENR(n); n != nil && err == nil && n.Seq() != 0 {
 			s.serverPool.Persist(n)
 		} else {
 			return 0
@@ -351,7 +351,7 @@ func (s *LightEthereum) Start() error {
 	// Regularly update shutdown marker
 	s.shutdownTracker.Start()
 
-	if s.udpEnabled && s.p2pServer.DiscV5 == nil {
+	if s.udpEnabled && s.p2pServer.DiscoveryV5() == nil {
 		s.udpEnabled = false
 		log.Error("Discovery v5 is not initialized")
 	}

les/enr_entry.go

@@ -56,7 +56,7 @@ func (eth *LightEthereum) setupDiscovery() (enode.Iterator, error) {
 
 	// Enable DHT.
 	if eth.udpEnabled {
-		it.AddSource(eth.p2pServer.DiscV5.RandomNodes())
+		it.AddSource(eth.p2pServer.DiscoveryV5().RandomNodes())
 	}
 
 	forkFilter := forkid.NewFilter(eth.blockchain)

les/server.go

@@ -196,8 +196,8 @@ func (s *LesServer) Start() error {
 	s.handler.start()
 	s.wg.Add(1)
 	go s.capacityManagement()
-	if s.p2pSrv.DiscV5 != nil {
-		s.p2pSrv.DiscV5.RegisterTalkHandler("vfx", s.vfluxServer.ServeEncoded)
+	if discv5 := s.p2pSrv.DiscoveryV5(); discv5 != nil {
+		discv5.RegisterTalkHandler("vfx", s.vfluxServer.ServeEncoded)
 	}
 	return nil
 }

node/api.go

@@ -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},
@@ -327,3 +331,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
+}

p2p/discover/common.go

@@ -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"
@@ -63,7 +67,7 @@ func (cfg Config) withDefaults() Config {
 	// Node table configuration:
 	if cfg.PingInterval == 0 {
 		// MODIFIED by Jakub Pajek (reduce disc traffic)
-		//cfg.PingInterval = 10 * time.Second
+		//cfg.PingInterval = 3 * time.Second
 		cfg.PingInterval = 30 * time.Second
 	}
 	if cfg.RefreshInterval == 0 {
@@ -95,9 +99,57 @@ type ReadPacket struct {
 	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)
+}
+
 func min(x, y int) int {
 	if x > y {
 		return y
 	}
 	return x
 }
+
+func min64(x, y int64) int64 {
+	if x > y {
+		return y
+	}
+	return x
+}

p2p/discover/lookup.go

@@ -133,7 +133,7 @@ func (it *lookup) startQueries() bool {
 }
 
 // MODIFIED by Jakub Pajek (reduce disc traffic)
-//func (it *lookup) slowdown() {
+// func (it *lookup) slowdown() {
 func (it *lookup) slowdown(d time.Duration) {
 	// MODIFIED by Jakub Pajek (reduce disc traffic)
 	//sleep := time.NewTimer(1 * time.Second)
@@ -148,32 +148,13 @@ func (it *lookup) slowdown(d time.Duration) {
 func (it *lookup) query(n *node, reply chan<- []*node) {
 	// ADDED by Jakub Pajek (reduce disc traffic)
 	it.slowdown(1 * time.Second)
-	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
 }

p2p/discover/node.go

@@ -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
-	livenessChecks uint      // how often liveness was checked
+	*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 {