feat(p2p): new p2p node representation #17643 #17753 #18963

cmd/bootnode/main.go

@@ -29,6 +29,7 @@ import (
 	"github.com/XinFinOrg/XDPoSChain/log"
 	"github.com/XinFinOrg/XDPoSChain/p2p/discover"
 	"github.com/XinFinOrg/XDPoSChain/p2p/discv5"
+	"github.com/XinFinOrg/XDPoSChain/p2p/enode"
 	"github.com/XinFinOrg/XDPoSChain/p2p/nat"
 	"github.com/XinFinOrg/XDPoSChain/p2p/netutil"
 )
@@ -37,7 +38,7 @@ func main() {
 	var (
 		listenAddr  = flag.String("addr", ":30301", "listen address")
 		genKey      = flag.String("genkey", "", "generate a node key")
-		writeAddr   = flag.Bool("writeaddress", false, "write out the node's pubkey hash and quit")
+		writeAddr   = flag.Bool("writeaddress", false, "write out the node's public key and quit")
 		nodeKeyFile = flag.String("nodekey", "", "private key filename")
 		nodeKeyHex  = flag.String("nodekeyhex", "", "private key as hex (for testing)")
 		natdesc     = flag.String("nat", "none", "port mapping mechanism (any|none|upnp|pmp|extip:<IP>)")
@@ -85,7 +86,7 @@ func main() {
 	}
 
 	if *writeAddr {
-		fmt.Printf("%v\n", discover.PubkeyID(&nodeKey.PublicKey))
+		fmt.Printf("%x\n", crypto.FromECDSAPub(&nodeKey.PublicKey)[1:])
 		os.Exit(0)
 	}
 
@@ -111,26 +112,38 @@ func main() {
 		if !realaddr.IP.IsLoopback() {
 			go nat.Map(natm, nil, "udp", realaddr.Port, realaddr.Port, "ethereum discovery")
 		}
-		// TODO: react to external IP changes over time.
 		if ext, err := natm.ExternalIP(); err == nil {
 			realaddr = &net.UDPAddr{IP: ext, Port: realaddr.Port}
 		}
 	}
 
+	printNotice(&nodeKey.PublicKey, *realaddr)
+
 	if *runv5 {
-		if _, err := discv5.ListenUDP(nodeKey, conn, realaddr, "", restrictList); err != nil {
+		if _, err := discv5.ListenUDP(nodeKey, conn, "", restrictList); err != nil {
 			utils.Fatalf("%v", err)
 		}
 	} else {
+		db, _ := enode.OpenDB("")
+		ln := enode.NewLocalNode(db, nodeKey)
 		cfg := discover.Config{
-			PrivateKey:   nodeKey,
-			AnnounceAddr: realaddr,
-			NetRestrict:  restrictList,
+			PrivateKey:  nodeKey,
+			NetRestrict: restrictList,
 		}
-		if _, err := discover.ListenUDP(conn, cfg); err != nil {
+		if _, err := discover.ListenUDP(conn, ln, cfg); err != nil {
 			utils.Fatalf("%v", err)
 		}
 	}
 
 	select {}
 }
+
+func printNotice(nodeKey *ecdsa.PublicKey, addr net.UDPAddr) {
+	if addr.IP.IsUnspecified() {
+		addr.IP = net.IP{127, 0, 0, 1}
+	}
+	n := enode.NewV4(nodeKey, addr.IP, 0, addr.Port)
+	fmt.Println(n.String())
+	fmt.Println("Note: you're using cmd/bootnode, a developer tool.")
+	fmt.Println("We recommend using a regular node as bootstrap node for production deployments.")
+}

cmd/p2psim/main.go

@@ -47,7 +47,7 @@ import (
 	"github.com/XinFinOrg/XDPoSChain/crypto"
 	"github.com/XinFinOrg/XDPoSChain/internal/flags"
 	"github.com/XinFinOrg/XDPoSChain/p2p"
-	"github.com/XinFinOrg/XDPoSChain/p2p/discover"
+	"github.com/XinFinOrg/XDPoSChain/p2p/enode"
 	"github.com/XinFinOrg/XDPoSChain/p2p/simulations"
 	"github.com/XinFinOrg/XDPoSChain/p2p/simulations/adapters"
 	"github.com/XinFinOrg/XDPoSChain/rpc"
@@ -300,7 +300,7 @@ func createNode(ctx *cli.Context) error {
 		if err != nil {
 			return err
 		}
-		config.ID = discover.PubkeyID(&privKey.PublicKey)
+		config.ID = enode.PubkeyToIDV4(&privKey.PublicKey)
 		config.PrivateKey = privKey
 	}
 	if services := ctx.String("services"); services != "" {

cmd/utils/flags.go

@@ -61,8 +61,8 @@ import (
 	"github.com/XinFinOrg/XDPoSChain/miner"
 	"github.com/XinFinOrg/XDPoSChain/node"
 	"github.com/XinFinOrg/XDPoSChain/p2p"
-	"github.com/XinFinOrg/XDPoSChain/p2p/discover"
 	"github.com/XinFinOrg/XDPoSChain/p2p/discv5"
+	"github.com/XinFinOrg/XDPoSChain/p2p/enode"
 	"github.com/XinFinOrg/XDPoSChain/p2p/nat"
 	"github.com/XinFinOrg/XDPoSChain/p2p/netutil"
 	"github.com/XinFinOrg/XDPoSChain/params"
@@ -938,19 +938,20 @@ func setAllowlistAndDenylistForPeers(ctx *cli.Context, cfg *p2p.Config) {
 	// setup allowlist for peers
 	if ctx.IsSet(PeersAllowlistFlag.Name) {
 		urls := SplitAndTrim(ctx.String(PeersAllowlistFlag.Name))
-		cfg.AllowPeers = make(map[discover.NodeID]struct{}, len(urls))
+		cfg.AllowPeers = make(map[enode.ID]struct{}, len(urls))
 		for _, url := range urls {
 			if url != "" {
-				node1, err1 := discover.HexID(url)
+				var node1 enode.ID
+				err1 := node1.UnmarshalText([]byte(url))
 				if err1 == nil {
 					cfg.AllowPeers[node1] = struct{}{}
 					log.Info("Add peer to allowlist", "id", node1)
 					continue
 				}
-				node2, err2 := discover.ParseNode(url)
+				node2, err2 := enode.ParseV4(url)
 				if err2 == nil {
-					cfg.AllowPeers[node2.ID] = struct{}{}
-					log.Info("Add peer to allowlist", "enode", url, "id", node2.ID)
+					cfg.AllowPeers[node2.ID()] = struct{}{}
+					log.Info("Add peer to allowlist", "enode", url, "id", node2.ID())
 					continue
 				}
 				log.Crit("Invalid peer id for allowlist", "url", url, "err1", err1, "err2", err2)
@@ -961,19 +962,20 @@ func setAllowlistAndDenylistForPeers(ctx *cli.Context, cfg *p2p.Config) {
 	// setup denylist for peers
 	if ctx.IsSet(PeersDenylistFlag.Name) {
 		urls := SplitAndTrim(ctx.String(PeersDenylistFlag.Name))
-		cfg.DenyPeers = make(map[discover.NodeID]struct{}, len(urls))
+		cfg.DenyPeers = make(map[enode.ID]struct{}, len(urls))
 		for _, url := range urls {
 			if url != "" {
-				node1, err1 := discover.HexID(url)
+				var node1 enode.ID
+				err1 := node1.UnmarshalText([]byte(url))
 				if err1 == nil {
 					cfg.DenyPeers[node1] = struct{}{}
 					log.Info("Add peer to denylist", "id", node1)
 					continue
 				}
-				node2, err2 := discover.ParseNode(url)
+				node2, err2 := enode.ParseV4(url)
 				if err2 == nil {
-					cfg.DenyPeers[node2.ID] = struct{}{}
-					log.Info("Add peer to denylist", "enode", url, "id", node2.ID)
+					cfg.DenyPeers[node2.ID()] = struct{}{}
+					log.Info("Add peer to denylist", "enode", url, "id", node2.ID())
 					continue
 				}
 				log.Crit("Invalid peer id for denylist", "url", url, "err1", err1, "err2", err2)
@@ -988,10 +990,10 @@ func removeDenylistedPeers(cfg *p2p.Config) {
 		return
 	}
 
-	filteredNodes := make([]*discover.Node, 0, len(cfg.BootstrapNodes))
+	filteredNodes := make([]*enode.Node, 0, len(cfg.BootstrapNodes))
 	for _, node := range cfg.BootstrapNodes {
-		if _, ok := cfg.DenyPeers[node.ID]; ok {
-			log.Info("Remove denylisted peer", "enode", node, "id", node.ID)
+		if _, ok := cfg.DenyPeers[node.ID()]; ok {
+			log.Info("Remove denylisted peer", "enode", node, "id", node.ID())
 			continue
 		}
 		filteredNodes = append(filteredNodes, node)
@@ -1027,11 +1029,11 @@ func setBootstrapNodes(ctx *cli.Context, cfg *p2p.Config) {
 	cfg.BootstrapNodes = mustParseBootnodes(urls)
 }
 
-func mustParseBootnodes(urls []string) []*discover.Node {
-	nodes := make([]*discover.Node, 0, len(urls))
+func mustParseBootnodes(urls []string) []*enode.Node {
+	nodes := make([]*enode.Node, 0, len(urls))
 	for _, url := range urls {
 		if url != "" {
-			node, err := discover.ParseNode(url)
+			node, err := enode.ParseV4(url)
 			if err != nil {
 				log.Crit("Bootstrap URL invalid", "enode", url, "err", err)
 				return nil

eth/handler.go

@@ -39,7 +39,7 @@ import (
 	"github.com/XinFinOrg/XDPoSChain/event"
 	"github.com/XinFinOrg/XDPoSChain/log"
 	"github.com/XinFinOrg/XDPoSChain/p2p"
-	"github.com/XinFinOrg/XDPoSChain/p2p/discover"
+	"github.com/XinFinOrg/XDPoSChain/p2p/enode"
 	"github.com/XinFinOrg/XDPoSChain/params"
 	"github.com/XinFinOrg/XDPoSChain/rlp"
 )
@@ -184,7 +184,7 @@ func NewProtocolManager(config *params.ChainConfig, mode downloader.SyncMode, ne
 			NodeInfo: func() interface{} {
 				return manager.NodeInfo()
 			},
-			PeerInfo: func(id discover.NodeID) interface{} {
+			PeerInfo: func(id enode.ID) interface{} {
 				if p := manager.peers.Peer(fmt.Sprintf("%x", id[:8])); p != nil {
 					return p.Info()
 				}
@@ -364,41 +364,38 @@ func (pm *ProtocolManager) handle(p *peer) error {
 		rw.Init(p.version)
 	}
 	// Register the peer locally
-	err := pm.peers.Register(p)
-	if err != nil && err != p2p.ErrAddPairPeer {
+	if err := pm.peers.Register(p); err != nil {
 		p.Log().Error("Ethereum peer registration failed", "err", err)
 		return err
 	}
 	defer pm.removePeer(p.id)
-	if err != p2p.ErrAddPairPeer {
-		// Register the peer in the downloader. If the downloader considers it banned, we disconnect
-		if err := pm.downloader.RegisterPeer(p.id, p.version, p); err != nil {
+	// Register the peer in the downloader. If the downloader considers it banned, we disconnect
+	if err := pm.downloader.RegisterPeer(p.id, p.version, p); err != nil {
+		return err
+	}
+	p.Log().Info("Register peer", "nodeid", p.ID().String(), "version", p.version, "addr", p.RemoteAddr())
+	// Propagate existing transactions. new transactions appearing
+	// after this will be sent via broadcasts.
+	pm.syncTransactions(p)
+
+	// If we're DAO hard-fork aware, validate any remote peer with regard to the hard-fork
+	if daoBlock := pm.chainconfig.DAOForkBlock; daoBlock != nil {
+		// Request the peer's DAO fork header for extra-data validation
+		if err := p.RequestHeadersByNumber(daoBlock.Uint64(), 1, 0, false); err != nil {
 			return err
 		}
-		p.Log().Info("Register peer", "nodeid", p.ID().String(), "version", p.version, "addr", p.RemoteAddr())
-		// Propagate existing transactions. new transactions appearing
-		// after this will be sent via broadcasts.
-		pm.syncTransactions(p)
-
-		// If we're DAO hard-fork aware, validate any remote peer with regard to the hard-fork
-		if daoBlock := pm.chainconfig.DAOForkBlock; daoBlock != nil {
-			// Request the peer's DAO fork header for extra-data validation
-			if err := p.RequestHeadersByNumber(daoBlock.Uint64(), 1, 0, false); err != nil {
-				return err
+		// Start a timer to disconnect if the peer doesn't reply in time
+		p.forkDrop = time.AfterFunc(daoChallengeTimeout, func() {
+			p.Log().Debug("Timed out DAO fork-check, dropping")
+			pm.removePeer(p.id)
+		})
+		// Make sure it's cleaned up if the peer dies off
+		defer func() {
+			if p.forkDrop != nil {
+				p.forkDrop.Stop()
+				p.forkDrop = nil
 			}
-			// Start a timer to disconnect if the peer doesn't reply in time
-			p.forkDrop = time.AfterFunc(daoChallengeTimeout, func() {
-				p.Log().Debug("Timed out DAO fork-check, dropping")
-				pm.removePeer(p.id)
-			})
-			// Make sure it's cleaned up if the peer dies off
-			defer func() {
-				if p.forkDrop != nil {
-					p.forkDrop.Stop()
-					p.forkDrop = nil
-				}
-			}()
-		}
+		}()
 	}
 	// main loop. handle incoming messages.
 	for {

eth/helper_test.go

@@ -40,7 +40,7 @@ import (
 	"github.com/XinFinOrg/XDPoSChain/ethdb"
 	"github.com/XinFinOrg/XDPoSChain/event"
 	"github.com/XinFinOrg/XDPoSChain/p2p"
-	"github.com/XinFinOrg/XDPoSChain/p2p/discover"
+	"github.com/XinFinOrg/XDPoSChain/p2p/enode"
 	"github.com/XinFinOrg/XDPoSChain/params"
 	"github.com/holiman/uint256"
 )
@@ -206,7 +206,7 @@ func newTestPeer(name string, version int, pm *ProtocolManager, shake bool) (*te
 	app, net := p2p.MsgPipe()
 
 	// Generate a random id and create the peer
-	var id discover.NodeID
+	var id enode.ID
 	rand.Read(id[:])
 
 	peer := pm.newPeer(version, p2p.NewPeer(id, name, nil), net)

eth/metrics.go

@@ -111,7 +111,33 @@ func (rw *meteredMsgReadWriter) ReadMsg() (p2p.Msg, error) {
 }
 
 func (rw *meteredMsgReadWriter) WriteMsg(msg p2p.Msg) error {
-	// Account for the data traffic
+	rw.meterOut(msg)
+	// Send the packet to the p2p layer
+	return rw.MsgReadWriter.WriteMsg(msg)
+}
+
+// Compile-time check that meteredMsgReadWriter forwards the priority lane.
+// Without this, a future refactor could silently drop the wrapper back to
+// MsgReadWriter and downgrade BFT consensus messages to the normal lane.
+var _ p2p.PriorityMsgWriter = (*meteredMsgReadWriter)(nil)
+
+// WriteMsgPriority implements p2p.PriorityMsgWriter so that consensus messages
+// retain their priority routing when the metrics wrapper is enabled. The
+// outbound counters are updated identically to WriteMsg before the message is
+// forwarded to the underlying writer's priority lane (falling back to WriteMsg
+// if the underlying writer does not support priorities).
+func (rw *meteredMsgReadWriter) WriteMsgPriority(msg p2p.Msg, high bool) error {
+	rw.meterOut(msg)
+	if pw, ok := rw.MsgReadWriter.(p2p.PriorityMsgWriter); ok {
+		return pw.WriteMsgPriority(msg, high)
+	}
+	return rw.MsgReadWriter.WriteMsg(msg)
+}
+
+// meterOut updates the outbound packet/traffic meters for msg. It is shared
+// between WriteMsg and WriteMsgPriority so both paths produce identical
+// metrics.
+func (rw *meteredMsgReadWriter) meterOut(msg p2p.Msg) {
 	packets, traffic := miscOutPacketsMeter, miscOutTrafficMeter
 	switch {
 	case msg.Code == BlockHeadersMsg:
@@ -133,7 +159,4 @@ func (rw *meteredMsgReadWriter) WriteMsg(msg p2p.Msg) error {
 	}
 	packets.Mark(1)
 	traffic.Mark(int64(msg.Size))
-
-	// Send the packet to the p2p layer
-	return rw.MsgReadWriter.WriteMsg(msg)
 }