From bda106bc16965b33316951f46a8a32f5e8f62af5 Mon Sep 17 00:00:00 2001 From: Daniel Liu <139250065@qq.com> Date: Thu, 2 Jul 2026 12:58:56 +0800 Subject: [PATCH] fix(p2p,eth): remove per-peer dual-connection mechanism, close #2351 #2359 XDPoSChain kept two independent RLPx/TCP connections per remote peer (a "primary" and a "pair"), a private divergence from upstream go-ethereum which allows only one connection per NodeID. This adds state-machine complexity (isPair, pairRw, PairPeer, ErrAddPairPeer, DiscPairPeerStop) without measured benefit, and the pair connection already carried both block-sync and BFT traffic, so it never isolated Vote/Timeout/SyncInfo from large BlockBodies/NodeData frames. Revert to a single connection per NodeID (upstream semantics) and route all eth messages, including the XDPoS v2 BFT messages, over that single connection using the existing upstream FIFO write scheduling. No write-priority lane is added; RLPx framing, message ids, capability strings and the handshake are unchanged, so the change is rolling-upgrade safe against legacy pair-capable nodes. Changes: - p2p/server.go: postHandshakeChecks rejects any duplicate NodeID with DiscAlreadyConnected; addPeer no longer stores a second connection. - p2p/peer.go: drop pairPeer field/methods and the DiscPairPeerStop cascade on Peer.run exit. - p2p/peer_error.go: remove ErrAddPairPeer; keep the DiscPairPeerStop enum slot (deprecated) to preserve on-the-wire numbering. - eth/peer.go: remove pairRw; all Send/Request helpers use p.rw; peerSet.Register reverts to one peer per id. - eth/handler.go: drop ErrAddPairPeer special-casing so every peer registers with the downloader and syncs transactions. Refs: https://github.com/XinFinOrg/XDPoSChain/issues/2359 --- eth/handler.go | 53 +++++++++++------------ eth/peer.go | 103 ++++++++------------------------------------- eth/peer_test.go | 22 ++++++++++ p2p/peer.go | 30 ------------- p2p/peer_error.go | 5 ++- p2p/peer_test.go | 28 ------------ p2p/server.go | 12 +----- p2p/server_test.go | 25 +++++++++++ 8 files changed, 95 insertions(+), 183 deletions(-) create mode 100644 eth/peer_test.go diff --git a/eth/handler.go b/eth/handler.go index a0e6ecfbca34..6ce236a7d9b2 100644 --- a/eth/handler.go +++ b/eth/handler.go @@ -351,41 +351,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.blockchain.Config().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.blockchain.Config().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 { diff --git a/eth/peer.go b/eth/peer.go index 4b3ab6fad4d3..a0be36675725 100644 --- a/eth/peer.go +++ b/eth/peer.go @@ -59,8 +59,7 @@ type peer struct { id string *p2p.Peer - rw p2p.MsgReadWriter - pairRw p2p.MsgReadWriter + rw p2p.MsgReadWriter version int // Protocol version negotiated forkDrop *time.Timer // Timed connection dropper if forks aren't validated in time @@ -262,59 +261,35 @@ func (p *peer) SendNewBlock(block *types.Block, td *big.Int) error { } p.knownBlocks.Add(block.Hash()) - if p.pairRw != nil { - return p2p.Send(p.pairRw, NewBlockMsg, []interface{}{block, td}) - } else { - return p2p.Send(p.rw, NewBlockMsg, []interface{}{block, td}) - } + return p2p.Send(p.rw, NewBlockMsg, []interface{}{block, td}) } // SendBlockHeaders sends a batch of block headers to the remote peer. func (p *peer) SendBlockHeaders(headers []*types.Header) error { - if p.pairRw != nil { - return p2p.Send(p.pairRw, BlockHeadersMsg, headers) - } else { - return p2p.Send(p.rw, BlockHeadersMsg, headers) - } + return p2p.Send(p.rw, BlockHeadersMsg, headers) } // SendBlockBodies sends a batch of block contents to the remote peer. func (p *peer) SendBlockBodies(bodies []*blockBody) error { - if p.pairRw != nil { - return p2p.Send(p.pairRw, BlockBodiesMsg, blockBodiesData(bodies)) - } else { - return p2p.Send(p.rw, BlockBodiesMsg, blockBodiesData(bodies)) - } + return p2p.Send(p.rw, BlockBodiesMsg, blockBodiesData(bodies)) } // SendBlockBodiesRLP sends a batch of block contents to the remote peer from // an already RLP encoded format. func (p *peer) SendBlockBodiesRLP(bodies []rlp.RawValue) error { - if p.pairRw != nil { - return p2p.Send(p.pairRw, BlockBodiesMsg, bodies) - } else { - return p2p.Send(p.rw, BlockBodiesMsg, bodies) - } + return p2p.Send(p.rw, BlockBodiesMsg, bodies) } // SendNodeDataRLP sends a batch of arbitrary internal data, corresponding to the // hashes requested. func (p *peer) SendNodeData(data [][]byte) error { - if p.pairRw != nil { - return p2p.Send(p.pairRw, NodeDataMsg, data) - } else { - return p2p.Send(p.rw, NodeDataMsg, data) - } + return p2p.Send(p.rw, NodeDataMsg, data) } // SendReceiptsRLP sends a batch of transaction receipts, corresponding to the // ones requested from an already RLP encoded format. func (p *peer) SendReceiptsRLP(receipts []rlp.RawValue) error { - if p.pairRw != nil { - return p2p.Send(p.pairRw, ReceiptsMsg, receipts) - } else { - return p2p.Send(p.rw, ReceiptsMsg, receipts) - } + return p2p.Send(p.rw, ReceiptsMsg, receipts) } func (p *peer) SendVote(vote *types.Vote) error { @@ -323,11 +298,7 @@ func (p *peer) SendVote(vote *types.Vote) error { } p.knownVote.Add(vote.Hash()) - if p.pairRw != nil { - return p2p.Send(p.pairRw, VoteMsg, vote) - } else { - return p2p.Send(p.rw, VoteMsg, vote) - } + return p2p.Send(p.rw, VoteMsg, vote) } /* @@ -341,11 +312,7 @@ func (p *peer) SendTimeout(timeout *types.Timeout) error { } p.knownTimeout.Add(timeout.Hash()) - if p.pairRw != nil { - return p2p.Send(p.pairRw, TimeoutMsg, timeout) - } else { - return p2p.Send(p.rw, TimeoutMsg, timeout) - } + return p2p.Send(p.rw, TimeoutMsg, timeout) } /* @@ -359,11 +326,7 @@ func (p *peer) SendSyncInfo(syncInfo *types.SyncInfo) error { } p.knownSyncInfo.Add(syncInfo.Hash()) - if p.pairRw != nil { - return p2p.Send(p.pairRw, SyncInfoMsg, syncInfo) - } else { - return p2p.Send(p.rw, SyncInfoMsg, syncInfo) - } + return p2p.Send(p.rw, SyncInfoMsg, syncInfo) } /* @@ -376,65 +339,41 @@ func (p *peer) AsyncSendSyncInfo() { // single header. It is used solely by the fetcher. func (p *peer) RequestOneHeader(hash common.Hash) error { p.Log().Debug("Fetching single header", "hash", hash) - if p.pairRw != nil { - return p2p.Send(p.pairRw, GetBlockHeadersMsg, &getBlockHeadersData{Origin: hashOrNumber{Hash: hash}, Amount: uint64(1), Skip: uint64(0), Reverse: false}) - } else { - return p2p.Send(p.rw, GetBlockHeadersMsg, &getBlockHeadersData{Origin: hashOrNumber{Hash: hash}, Amount: uint64(1), Skip: uint64(0), Reverse: false}) - } + return p2p.Send(p.rw, GetBlockHeadersMsg, &getBlockHeadersData{Origin: hashOrNumber{Hash: hash}, Amount: uint64(1), Skip: uint64(0), Reverse: false}) } // RequestHeadersByHash fetches a batch of blocks' headers corresponding to the // specified header query, based on the hash of an origin block. func (p *peer) RequestHeadersByHash(origin common.Hash, amount int, skip int, reverse bool) error { p.Log().Debug("Fetching batch of headers", "count", amount, "fromhash", origin, "skip", skip, "reverse", reverse) - if p.pairRw != nil { - return p2p.Send(p.pairRw, GetBlockHeadersMsg, &getBlockHeadersData{Origin: hashOrNumber{Hash: origin}, Amount: uint64(amount), Skip: uint64(skip), Reverse: reverse}) - } else { - return p2p.Send(p.rw, GetBlockHeadersMsg, &getBlockHeadersData{Origin: hashOrNumber{Hash: origin}, Amount: uint64(amount), Skip: uint64(skip), Reverse: reverse}) - } + return p2p.Send(p.rw, GetBlockHeadersMsg, &getBlockHeadersData{Origin: hashOrNumber{Hash: origin}, Amount: uint64(amount), Skip: uint64(skip), Reverse: reverse}) } // RequestHeadersByNumber fetches a batch of blocks' headers corresponding to the // specified header query, based on the number of an origin block. func (p *peer) RequestHeadersByNumber(origin uint64, amount int, skip int, reverse bool) error { p.Log().Debug("Fetching batch of headers", "count", amount, "fromnum", origin, "skip", skip, "reverse", reverse) - if p.pairRw != nil { - return p2p.Send(p.pairRw, GetBlockHeadersMsg, &getBlockHeadersData{Origin: hashOrNumber{Number: origin}, Amount: uint64(amount), Skip: uint64(skip), Reverse: reverse}) - } else { - return p2p.Send(p.rw, GetBlockHeadersMsg, &getBlockHeadersData{Origin: hashOrNumber{Number: origin}, Amount: uint64(amount), Skip: uint64(skip), Reverse: reverse}) - } + return p2p.Send(p.rw, GetBlockHeadersMsg, &getBlockHeadersData{Origin: hashOrNumber{Number: origin}, Amount: uint64(amount), Skip: uint64(skip), Reverse: reverse}) } // RequestBodies fetches a batch of blocks' bodies corresponding to the hashes // specified. func (p *peer) RequestBodies(hashes []common.Hash) error { p.Log().Debug("Fetching batch of block bodies", "count", len(hashes)) - if p.pairRw != nil { - return p2p.Send(p.pairRw, GetBlockBodiesMsg, hashes) - } else { - return p2p.Send(p.rw, GetBlockBodiesMsg, hashes) - } + return p2p.Send(p.rw, GetBlockBodiesMsg, hashes) } // RequestNodeData fetches a batch of arbitrary data from a node's known state // data, corresponding to the specified hashes. func (p *peer) RequestNodeData(hashes []common.Hash) error { p.Log().Debug("Fetching batch of state data", "count", len(hashes)) - if p.pairRw != nil { - return p2p.Send(p.pairRw, GetNodeDataMsg, hashes) - } else { - return p2p.Send(p.rw, GetNodeDataMsg, hashes) - } + return p2p.Send(p.rw, GetNodeDataMsg, hashes) } // RequestReceipts fetches a batch of transaction receipts from a remote node. func (p *peer) RequestReceipts(hashes []common.Hash) error { p.Log().Debug("Fetching batch of receipts", "count", len(hashes)) - if p.pairRw != nil { - return p2p.Send(p.pairRw, GetReceiptsMsg, hashes) - } else { - return p2p.Send(p.rw, GetReceiptsMsg, hashes) - } + return p2p.Send(p.rw, GetReceiptsMsg, hashes) } // Handshake executes the eth protocol handshake, negotiating version number, @@ -530,14 +469,8 @@ func (ps *peerSet) Register(p *peer) error { if ps.closed { return errClosed } - if existPeer, ok := ps.peers[p.id]; ok { - if existPeer.pairRw != nil { - return errAlreadyRegistered - } - existPeer.SetPairPeer(p.Peer) - existPeer.pairRw = p.rw - p.SetPairPeer(existPeer.Peer) - return p2p.ErrAddPairPeer + if _, ok := ps.peers[p.id]; ok { + return errAlreadyRegistered } ps.peers[p.id] = p return nil diff --git a/eth/peer_test.go b/eth/peer_test.go new file mode 100644 index 000000000000..95c43062db15 --- /dev/null +++ b/eth/peer_test.go @@ -0,0 +1,22 @@ +package eth + +import "testing" + +func TestPeerSetRegisterRejectsDuplicateID(t *testing.T) { + peers := newPeerSet() + first := &peer{id: "dup"} + second := &peer{id: "dup"} + + if err := peers.Register(first); err != nil { + t.Fatalf("first register failed: %v", err) + } + if err := peers.Register(second); err != errAlreadyRegistered { + t.Fatalf("second register error mismatch: got %v want %v", err, errAlreadyRegistered) + } + if peers.Len() != 1 { + t.Fatalf("peer set size mismatch: got %d want 1", peers.Len()) + } + if got := peers.Peer("dup"); got != first { + t.Fatalf("registered peer replaced: got %p want %p", got, first) + } +} diff --git a/p2p/peer.go b/p2p/peer.go index 1c813666e12c..5f6e16f779c2 100644 --- a/p2p/peer.go +++ b/p2p/peer.go @@ -117,9 +117,6 @@ type Peer struct { // events receives message send / receive events if set events *event.Feed - - pairPeerMu sync.RWMutex - pairPeer *Peer } // NewPeer returns a peer for testing purposes. @@ -202,30 +199,6 @@ func (p *Peer) Log() log.Logger { return p.log } -func (p *Peer) PairPeer() *Peer { - p.pairPeerMu.RLock() - defer p.pairPeerMu.RUnlock() - - return p.pairPeer -} - -func (p *Peer) SetPairPeer(pair *Peer) { - p.pairPeerMu.Lock() - p.pairPeer = pair - p.pairPeerMu.Unlock() -} - -func (p *Peer) ClearPairPeer(pair *Peer) bool { - p.pairPeerMu.Lock() - defer p.pairPeerMu.Unlock() - - if p.pairPeer != pair { - return false - } - p.pairPeer = nil - return true -} - func (p *Peer) run() (remoteRequested bool, err error) { var ( writeStart = make(chan struct{}, 1) @@ -271,9 +244,6 @@ loop: close(p.closed) p.rw.close(reason) p.wg.Wait() - if pairPeer := p.PairPeer(); pairPeer != nil { - go pairPeer.Disconnect(DiscPairPeerStop) - } return remoteRequested, err } diff --git a/p2p/peer_error.go b/p2p/peer_error.go index 9c3e58261e45..fe1a8b7770af 100644 --- a/p2p/peer_error.go +++ b/p2p/peer_error.go @@ -54,8 +54,6 @@ func (pe *peerError) Error() string { var errProtocolReturned = errors.New("protocol returned") -var ErrAddPairPeer = errors.New("add a pair peer") - type DiscReason uint const ( @@ -71,6 +69,9 @@ const ( DiscUnexpectedIdentity DiscSelf DiscReadTimeout + // Deprecated: the per-peer pair connection mechanism was removed (issue #2359). + // The enum slot is kept to preserve on-the-wire numbering of the values below + // for rolling-upgrade compatibility. Do not reuse. DiscPairPeerStop DiscNonAllowlistedPeer DiscDenylistedPeer diff --git a/p2p/peer_test.go b/p2p/peer_test.go index a1de73ceb5e2..f1d4fe225371 100644 --- a/p2p/peer_test.go +++ b/p2p/peer_test.go @@ -239,34 +239,6 @@ func TestPeerDisconnectRace(t *testing.T) { } } -func TestPeerRunDisconnectsPairPeer(t *testing.T) { - closer, _, peer, errc := testPeer(nil) - defer closer() - - pairPeer := &Peer{ - disc: make(chan DiscReason, 1), - closed: make(chan struct{}), - } - peer.SetPairPeer(pairPeer) - - closer() - - select { - case <-errc: - case <-time.After(2 * time.Second): - t.Fatal("peer did not stop") - } - - select { - case reason := <-pairPeer.disc: - if reason != DiscPairPeerStop { - t.Fatalf("unexpected pair disconnect reason: got %v want %v", reason, DiscPairPeerStop) - } - case <-time.After(2 * time.Second): - t.Fatal("pair peer was not disconnected") - } -} - func TestNewPeer(t *testing.T) { name := "nodename" caps := []Cap{{"foo", 2}, {"bar", 3}} diff --git a/p2p/server.go b/p2p/server.go index 24f1e1a22e8a..8a824ec9ee5b 100644 --- a/p2p/server.go +++ b/p2p/server.go @@ -777,11 +777,7 @@ running: if err == nil { // The handshakes are done and it passed all checks. p := srv.launchPeer(c) - if peers[c.node.ID()] != nil { - peers[c.node.ID()].SetPairPeer(p) - } else { - peers[c.node.ID()] = p - } + peers[c.node.ID()] = p srv.log.Debug("Adding p2p peer", "peercount", len(peers), "id", p.ID(), "conn", c.flags, "addr", p.RemoteAddr(), "name", truncateName(c.name)) srv.dialsched.peerAdded(c) if p.Inbound() { @@ -837,11 +833,7 @@ func (srv *Server) postHandshakeChecks(peers map[enode.ID]*Peer, inboundCount in case !c.is(trustedConn) && c.is(inboundConn) && inboundCount >= srv.maxInboundConns(): return DiscTooManyPeers case peers[c.node.ID()] != nil: - existingPeer := peers[c.node.ID()] - if existingPeer.PairPeer() != nil { - return DiscAlreadyConnected - } - return nil + return DiscAlreadyConnected case c.node.ID() == srv.localnode.ID(): return DiscSelf default: diff --git a/p2p/server_test.go b/p2p/server_test.go index d5b92b7e0527..ed3fa88e3a08 100644 --- a/p2p/server_test.go +++ b/p2p/server_test.go @@ -297,6 +297,31 @@ func TestServerAtCap(t *testing.T) { t.Error("Server did not set trusted flag") } } + +func TestPostHandshakeChecksRejectsDuplicateNodeID(t *testing.T) { + id := randomID() + srv := &Server{Config: Config{MaxPeers: 10}} + peers := map[enode.ID]*Peer{id: {}} + + tests := []struct { + name string + flags connFlag + }{ + {name: "inbound", flags: inboundConn}, + {name: "trusted inbound", flags: inboundConn | trustedConn}, + {name: "dialed", flags: dynDialedConn}, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + c := &conn{node: newNode(id, ""), flags: tc.flags} + if err := srv.postHandshakeChecks(peers, 0, c); err != DiscAlreadyConnected { + t.Fatalf("unexpected error: got %v want %v", err, DiscAlreadyConnected) + } + }) + } +} + func TestServerPeerLimits(t *testing.T) { srvkey := newkey() clientkey := newkey()