forks/go-ethereum
1
0
Fork 1
mirror of https://github.com/ethereum/go-ethereum.git synced 2026-07-19 03:10:48 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 1

cmd/devp2p, p2p: fix discovery RPC listener issues

Browse source 
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.
This commit is contained in:
Chase Wright 2026-07-10 10:22:41 -05:00
parent 95a3b35e97
commit 9b5737150a
No known key found for this signature in database
GPG key ID: 7EA41EDCB1BE04A7
6 changed files with 70 additions and 113 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

52
cmd/devp2p/discoverycmd.go
View file

@ -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},
})
}

29
cmd/devp2p/discv4cmd.go
View file

@ -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 {

30
cmd/devp2p/discv5cmd.go
View file

@ -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 {

19
cmd/devp2p/main.go
View file

@ -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)

25
p2p/discover/common.go
View file

@ -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

28
p2p/server.go
View file

@ -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.