From 9b5737150a503607ac5e4b40635dbe6db1268c86 Mon Sep 17 00:00:00 2001 From: Chase Wright Date: Fri, 10 Jul 2026 10:22:41 -0500 Subject: [PATCH] cmd/devp2p, p2p: fix discovery RPC listener issues Serve the RPC handler directly instead of via http.DefaultServeMux, which also exposed /debug/pprof on the RPC port. lookupRandom now closes its iterator and stops when the client disconnects, instead of hanging forever on nodes with an empty table and leaking a context per call. Results go through enode.ReadNodes: deduplicated (may return fewer than n) and empty responses marshal as [] instead of null. The discv4 issues predate this branch; fixed here because the combined listener exposes them. Also export discover.SharedUDPConn and use it from p2p.Server and the combined listener instead of keeping two diverging copies. --- cmd/devp2p/discoverycmd.go | 52 ++++++-------------------------------- cmd/devp2p/discv4cmd.go | 29 ++++++--------------- cmd/devp2p/discv5cmd.go | 30 +++++++--------------- cmd/devp2p/main.go | 19 ++++++++++++++ p2p/discover/common.go | 25 ++++++++++++++++++ p2p/server.go | 28 +------------------- 6 files changed, 70 insertions(+), 113 deletions(-) diff --git a/cmd/devp2p/discoverycmd.go b/cmd/devp2p/discoverycmd.go index 9872fb3697..fb6dc0df08 100644 --- a/cmd/devp2p/discoverycmd.go +++ b/cmd/devp2p/discoverycmd.go @@ -17,16 +17,10 @@ package main import ( - "errors" "fmt" - "net" - "net/http" - "net/netip" "slices" - "github.com/ethereum/go-ethereum/log" "github.com/ethereum/go-ethereum/p2p/discover" - "github.com/ethereum/go-ethereum/rpc" "github.com/urfave/cli/v2" ) @@ -55,13 +49,13 @@ func discoveryListen(ctx *cli.Context) error { // v4 is the primary listener on the real socket and forwards packets it // can't parse to v5 via the unhandled channel. unhandled := make(chan discover.ReadPacket, 100) - config.Unhandled = unhandled - v4, err := discover.ListenV4(socket, ln, config) + v4cfg := config + v4cfg.Unhandled = unhandled + v4, err := discover.ListenV4(socket, ln, v4cfg) if err != nil { exit(err) } - config.Unhandled = nil - v5, err := discover.ListenV5(&sharedUDPConn{socket, unhandled}, ln, config) + v5, err := discover.ListenV5(&discover.SharedUDPConn{UDPConn: socket, Unhandled: unhandled}, ln, config) if err != nil { exit(err) } @@ -75,38 +69,8 @@ func discoveryListen(ctx *cli.Context) error { // v4 and v5 share the same local node, so this is the single record both announce. fmt.Println(ln.Node()) - httpAddr := ctx.String(httpAddrFlag.Name) - if httpAddr == "" { - select {} - } - - srv := rpc.NewServer() - srv.RegisterName("discv4", &discv4API{v4}) - srv.RegisterName("discv5", &discv5API{v5}) - log.Info("Starting RPC API server", "addr", httpAddr) - http.DefaultServeMux.Handle("/", srv) - httpsrv := http.Server{Addr: httpAddr, Handler: http.DefaultServeMux} - return httpsrv.ListenAndServe() -} - -// sharedUDPConn implements a shared connection, identical to the one used by -// p2p.Server: reads return packets the primary v4 listener found unprocessable -// and forwarded on the unhandled channel. -type sharedUDPConn struct { - *net.UDPConn - unhandled chan discover.ReadPacket -} - -func (s *sharedUDPConn) ReadFromUDPAddrPort(b []byte) (n int, addr netip.AddrPort, err error) { - packet, ok := <-s.unhandled - if !ok { - return 0, netip.AddrPort{}, errors.New("connection was closed") - } - l := min(len(packet.Data), len(b)) - copy(b[:l], packet.Data[:l]) - return l, packet.Addr, nil -} - -func (s *sharedUDPConn) Close() error { - return nil + return runRPCServer(ctx.String(httpAddrFlag.Name), map[string]any{ + "discv4": &discv4API{v4}, + "discv5": &discv5API{v5}, + }) } diff --git a/cmd/devp2p/discv4cmd.go b/cmd/devp2p/discv4cmd.go index 98b04311b4..146e8be1e9 100644 --- a/cmd/devp2p/discv4cmd.go +++ b/cmd/devp2p/discv4cmd.go @@ -17,10 +17,10 @@ package main import ( + "context" "errors" "fmt" "net" - "net/http" "slices" "strconv" "strings" @@ -29,11 +29,9 @@ import ( "github.com/ethereum/go-ethereum/cmd/devp2p/internal/v4test" "github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/crypto" - "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" ) @@ -176,19 +174,7 @@ func discv4Listen(ctx *cli.Context) error { 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() + return runRPCServer(ctx.String(httpAddrFlag.Name), map[string]any{"discv4": &discv4API{disc}}) } func discv4RequestRecord(ctx *cli.Context) error { @@ -428,13 +414,14 @@ type discv4API struct { host *discover.UDPv4 } -func (api *discv4API) LookupRandom(n int) (ns []*enode.Node) { +func (api *discv4API) LookupRandom(ctx context.Context, n int) []*enode.Node { it := api.host.RandomNodes() defer it.Close() - for len(ns) < n && it.Next() { - ns = append(ns, it.Node()) - } - return ns + go func() { + <-ctx.Done() + it.Close() + }() + return enode.ReadNodes(it, n) } func (api *discv4API) Buckets() [][]discover.BucketNode { diff --git a/cmd/devp2p/discv5cmd.go b/cmd/devp2p/discv5cmd.go index 2bbc7c437b..5336e65678 100644 --- a/cmd/devp2p/discv5cmd.go +++ b/cmd/devp2p/discv5cmd.go @@ -17,18 +17,16 @@ package main import ( + "context" "errors" "fmt" - "net/http" "slices" "time" "github.com/ethereum/go-ethereum/cmd/devp2p/internal/v5test" "github.com/ethereum/go-ethereum/common" - "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/rpc" "github.com/urfave/cli/v2" ) @@ -144,19 +142,7 @@ func discv5Listen(ctx *cli.Context) error { fmt.Println(disc.Self()) - httpAddr := ctx.String(httpAddrFlag.Name) - if httpAddr == "" { - // Non-HTTP mode. - select {} - } - - api := &discv5API{disc} - log.Info("Starting RPC API server", "addr", httpAddr) - srv := rpc.NewServer() - srv.RegisterName("discv5", api) - http.DefaultServeMux.Handle("/", srv) - httpsrv := http.Server{Addr: httpAddr, Handler: http.DefaultServeMux} - return httpsrv.ListenAndServe() + return runRPCServer(ctx.String(httpAddrFlag.Name), map[string]any{"discv5": &discv5API{disc}}) } // startV5 starts an ephemeral discovery v5 node. @@ -174,12 +160,14 @@ type discv5API struct { host *discover.UDPv5 } -func (api *discv5API) LookupRandom(n int) (ns []*enode.Node) { +func (api *discv5API) LookupRandom(ctx context.Context, n int) []*enode.Node { it := api.host.RandomNodes() - for len(ns) < n && it.Next() { - ns = append(ns, it.Node()) - } - return ns + defer it.Close() + go func() { + <-ctx.Done() + it.Close() + }() + return enode.ReadNodes(it, n) } func (api *discv5API) Self() *enode.Node { diff --git a/cmd/devp2p/main.go b/cmd/devp2p/main.go index f33a63962e..b6df747022 100644 --- a/cmd/devp2p/main.go +++ b/cmd/devp2p/main.go @@ -18,11 +18,14 @@ package main import ( "fmt" + "net/http" "os" "github.com/ethereum/go-ethereum/internal/debug" "github.com/ethereum/go-ethereum/internal/flags" + "github.com/ethereum/go-ethereum/log" "github.com/ethereum/go-ethereum/p2p/enode" + "github.com/ethereum/go-ethereum/rpc" "github.com/urfave/cli/v2" ) @@ -93,6 +96,22 @@ func getNodeArg(ctx *cli.Context) *enode.Node { return n } +// runRPCServer serves the given RPC APIs on httpAddr; if httpAddr is empty, it blocks forever without serving. +func runRPCServer(httpAddr string, apis map[string]any) error { + if httpAddr == "" { + select {} + } + srv := rpc.NewServer() + for name, api := range apis { + if err := srv.RegisterName(name, api); err != nil { + return err + } + } + log.Info("Starting RPC API server", "addr", httpAddr) + httpsrv := http.Server{Addr: httpAddr, Handler: srv} + return httpsrv.ListenAndServe() +} + func exit(err interface{}) { if err == nil { os.Exit(0) diff --git a/p2p/discover/common.go b/p2p/discover/common.go index 5513afd54d..78146391ea 100644 --- a/p2p/discover/common.go +++ b/p2p/discover/common.go @@ -21,6 +21,7 @@ import ( "crypto/ecdsa" crand "crypto/rand" "encoding/binary" + "errors" "iter" "math/rand" "net" @@ -105,6 +106,30 @@ type ReadPacket struct { Addr netip.AddrPort } +// SharedUDPConn implements a shared connection. Write sends messages to the underlying +// connection while read returns messages that were found unprocessable and sent to the +// Unhandled channel by the primary listener. +type SharedUDPConn struct { + *net.UDPConn + Unhandled chan ReadPacket +} + +// ReadFromUDPAddrPort implements UDPConn. +func (s *SharedUDPConn) ReadFromUDPAddrPort(b []byte) (n int, addr netip.AddrPort, err error) { + packet, ok := <-s.Unhandled + if !ok { + return 0, netip.AddrPort{}, errors.New("connection was closed") + } + l := min(len(packet.Data), len(b)) + copy(b[:l], packet.Data[:l]) + return l, packet.Addr, nil +} + +// Close implements UDPConn. It does not close the underlying connection. +func (s *SharedUDPConn) Close() error { + return nil +} + type randomSource interface { Intn(int) int Int63n(int64) int64 diff --git a/p2p/server.go b/p2p/server.go index 6d2323f9ce..e12ca5b3c2 100644 --- a/p2p/server.go +++ b/p2p/server.go @@ -330,32 +330,6 @@ func (srv *Server) Stop() { srv.loopWG.Wait() } -// sharedUDPConn implements a shared connection. Write sends messages to the underlying connection while read returns -// messages that were found unprocessable and sent to the unhandled channel by the primary listener. -type sharedUDPConn struct { - *net.UDPConn - unhandled chan discover.ReadPacket -} - -// ReadFromUDPAddrPort implements discover.UDPConn -func (s *sharedUDPConn) ReadFromUDPAddrPort(b []byte) (n int, addr netip.AddrPort, err error) { - packet, ok := <-s.unhandled - if !ok { - return 0, netip.AddrPort{}, errors.New("connection was closed") - } - l := len(packet.Data) - if l > len(b) { - l = len(b) - } - copy(b[:l], packet.Data[:l]) - return l, packet.Addr, nil -} - -// Close implements discover.UDPConn -func (s *sharedUDPConn) Close() error { - return nil -} - // Start starts running the server. // Servers can not be re-used after stopping. func (srv *Server) Start() (err error) { @@ -463,7 +437,7 @@ func (srv *Server) setupDiscovery() error { // connection, so v5 can read unhandled messages from v4. if srv.Config.DiscoveryV4 && srv.Config.DiscoveryV5 { unhandled = make(chan discover.ReadPacket, 100) - sconn = &sharedUDPConn{conn, unhandled} + sconn = &discover.SharedUDPConn{UDPConn: conn, Unhandled: unhandled} } // Start discovery services.