Merge pull request #1267 from maticnetwork/psp-p2p-upstream

p2p: cherry-pick commits from geth for peering issues

Author: pratikspatil024

cmd/devp2p/discv4cmd.go

@@ -20,6 +20,7 @@ import (
 	"errors"
 	"fmt"
 	"net"
+	"net/http"
 	"strconv"
 	"strings"
 	"time"
@@ -30,9 +31,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"
 )
 
 var (
@@ -46,6 +49,7 @@ var (
 			discv4ResolveJSONCommand,
 			discv4CrawlCommand,
 			discv4TestCommand,
+			discv4ListenCommand,
 		},
 	}
 	discv4PingCommand = &cli.Command{
@@ -76,6 +80,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",
@@ -132,6 +144,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{
@@ -158,6 +174,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("", 0, 0)
+	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)
@@ -401,3 +438,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()
+}

cmd/devp2p/internal/v4test/framework.go

@@ -128,7 +128,7 @@ func (te *testenv) localEndpoint(c net.PacketConn) v4wire.Endpoint {
 }
 
 func (te *testenv) remoteEndpoint() v4wire.Endpoint {
-	return v4wire.NewEndpoint(te.remoteAddr, 0)
+	return v4wire.NewEndpoint(te.remoteAddr.AddrPort(), 0)
 }
 
 func contains(ns []v4wire.Node, key v4wire.Pubkey) bool {

internal/testlog/testlog.go

@@ -55,7 +55,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 {

node/api.go

@@ -27,6 +27,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"
 )
@@ -40,6 +41,9 @@ func (n *Node) apis() []rpc.API {
 		}, {
 			Namespace: "debug",
 			Service:   debug.Handler,
+		}, {
+			Namespace: "debug",
+			Service:   &p2pDebugAPI{n},
 		}, {
 			Namespace: "web3",
 			Service:   &web3API{n},
@@ -477,3 +481,16 @@ func (api *adminAPI) SetHttpExecutionPoolSize(n int) *ExecutionPoolSize {
 
 	return api.GetExecutionPoolSize()
 }
+
+// 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/dial.go

@@ -25,6 +25,7 @@ import (
 	mrand "math/rand"
 	"net"
 	"sync"
+	"sync/atomic"
 	"time"
 
 	"github.com/ethereum/go-ethereum/common/mclock"
@@ -254,7 +255,7 @@ loop:
 			}
 
 		case task := <-d.doneCh:
-			id := task.dest.ID()
+			id := task.dest().ID()
 			delete(d.dialing, id)
 			d.updateStaticPool(id)
 			d.doneSinceLastLog++
@@ -431,7 +432,7 @@ func (d *dialScheduler) startStaticDials(n int) (started int) {
 // updateStaticPool attempts to move the given static dial back into staticPool.
 func (d *dialScheduler) updateStaticPool(id enode.ID) {
 	task, ok := d.static[id]
-	if ok && task.staticPoolIndex < 0 && d.checkDial(task.dest) == nil {
+	if ok && task.staticPoolIndex < 0 && d.checkDial(task.dest()) == nil {
 		d.addToStaticPool(task)
 	}
 }
@@ -459,11 +460,11 @@ func (d *dialScheduler) removeFromStaticPool(idx int) {
 
 // startDial runs the given dial task in a separate goroutine.
 func (d *dialScheduler) startDial(task *dialTask) {
-	d.log.Trace("Starting p2p dial", "id", task.dest.ID(), "ip", task.dest.IP(), "flag", task.flags)
-	hkey := string(task.dest.ID().Bytes())
+	node := task.dest()
+	d.log.Trace("Starting p2p dial", "id", node.ID(), "ip", node.IP(), "flag", task.flags)
+	hkey := string(node.ID().Bytes())
 	d.history.add(hkey, d.clock.Now().Add(dialHistoryExpiration))
-	d.dialing[task.dest.ID()] = task
-
+	d.dialing[node.ID()] = task
 	go func() {
 		task.run(d)
 		d.doneCh <- task
@@ -474,39 +475,46 @@ func (d *dialScheduler) startDial(task *dialTask) {
 type dialTask struct {
 	staticPoolIndex int
 	flags           connFlag
+
 	// These fields are private to the task and should not be
 	// accessed by dialScheduler while the task is running.
-	dest         *enode.Node
+	destPtr      atomic.Pointer[enode.Node]
 	lastResolved mclock.AbsTime
 	resolveDelay time.Duration
 }
 
 func newDialTask(dest *enode.Node, flags connFlag) *dialTask {
-	return &dialTask{dest: dest, flags: flags, staticPoolIndex: -1}
+	t := &dialTask{flags: flags, staticPoolIndex: -1}
+	t.destPtr.Store(dest)
+	return t
 }
 
 type dialError struct {
 	error
 }
 
+func (t *dialTask) dest() *enode.Node {
+	return t.destPtr.Load()
+}
+
 func (t *dialTask) run(d *dialScheduler) {
 	if t.needResolve() && !t.resolve(d) {
 		return
 	}
 
-	err := t.dial(d, t.dest)
+	err := t.dial(d, t.dest())
 	if err != nil {
 		// For static nodes, resolve one more time if dialing fails.
 		if _, ok := err.(*dialError); ok && t.flags&staticDialedConn != 0 {
 			if t.resolve(d) {
-				t.dial(d, t.dest)
+				t.dial(d, t.dest())
 			}
 		}
 	}
 }
 
 func (t *dialTask) needResolve() bool {
-	return t.flags&staticDialedConn != 0 && t.dest.IP() == nil
+	return t.flags&staticDialedConn != 0 && t.dest().IP() == nil
 }
 
 // resolve attempts to find the current endpoint for the destination
@@ -528,42 +536,41 @@ func (t *dialTask) resolve(d *dialScheduler) bool {
 		return false
 	}
 
-	resolved := d.resolver.Resolve(t.dest)
+	node := t.dest()
+	resolved := d.resolver.Resolve(node)
 	t.lastResolved = d.clock.Now()
 
 	if resolved == nil {
 		t.resolveDelay *= 2
 		if t.resolveDelay > maxResolveDelay {
 			t.resolveDelay = maxResolveDelay
 		}
-
-		d.log.Debug("Resolving node failed", "id", t.dest.ID(), "newdelay", t.resolveDelay)
-
+		d.log.Debug("Resolving node failed", "id", node.ID(), "newdelay", t.resolveDelay)
 		return false
 	}
 	// The node was found.
 	t.resolveDelay = initialResolveDelay
-	t.dest = resolved
-	d.log.Debug("Resolved node", "id", t.dest.ID(), "addr", &net.TCPAddr{IP: t.dest.IP(), Port: t.dest.TCP()})
-
+	t.destPtr.Store(resolved)
+	d.log.Debug("Resolved node", "id", resolved.ID(), "addr", &net.TCPAddr{IP: resolved.IP(), Port: resolved.TCP()})
 	return true
 }
 
 // dial performs the actual connection attempt.
 func (t *dialTask) dial(d *dialScheduler, dest *enode.Node) error {
 	dialMeter.Mark(1)
-	fd, err := d.dialer.Dial(d.ctx, t.dest)
+	fd, err := d.dialer.Dial(d.ctx, dest)
 	if err != nil {
-		d.log.Trace("Dial error", "id", t.dest.ID(), "addr", nodeAddr(t.dest), "conn", t.flags, "err", cleanupDialErr(err))
+		d.log.Trace("Dial error", "id", dest.ID(), "addr", nodeAddr(dest), "conn", t.flags, "err", cleanupDialErr(err))
 		dialConnectionError.Mark(1)
 		return &dialError{err}
 	}
 	return d.setupFunc(newMeteredConn(fd), t.flags, dest)
 }
 
 func (t *dialTask) String() string {
-	id := t.dest.ID()
-	return fmt.Sprintf("%v %x %v:%d", t.flags, id[:8], t.dest.IP(), t.dest.TCP())
+	node := t.dest()
+	id := node.ID()
+	return fmt.Sprintf("%v %x %v:%d", t.flags, id[:8], node.IP(), node.TCP())
 }
 
 func cleanupDialErr(err error) error {

p2p/discover/common.go

@@ -18,7 +18,12 @@ package discover
 
 import (
 	"crypto/ecdsa"
+	crand "crypto/rand"
+	"encoding/binary"
+	"math/rand"
 	"net"
+	"net/netip"
+	"sync"
 	"time"
 
 	"github.com/ethereum/go-ethereum/common/mclock"
@@ -30,8 +35,8 @@ import (
 
 // UDPConn is a network connection on which discovery can operate.
 type UDPConn interface {
-	ReadFromUDP(b []byte) (n int, addr *net.UDPAddr, err error)
-	WriteToUDP(b []byte, addr *net.UDPAddr) (n int, err error)
+	ReadFromUDPAddrPort(b []byte) (n int, addr netip.AddrPort, err error)
+	WriteToUDPAddrPort(b []byte, addr netip.AddrPort) (n int, err error)
 	Close() error
 	LocalAddr() net.Addr
 }
@@ -62,7 +67,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
@@ -93,13 +98,46 @@ func ListenUDP(c UDPConn, ln *enode.LocalNode, cfg Config) (*UDPv4, error) {
 // channel if configured.
 type ReadPacket struct {
 	Data []byte
-	Addr *net.UDPAddr
+	Addr netip.AddrPort
 }
 
-func min(x, y int) int {
-	if x > y {
-		return y
-	}
+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)
+}
 
-	return x
+func (r *reseedingRandom) Shuffle(n int, swap func(i, j int)) {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	r.cur.Shuffle(n, swap)
 }