refactor(p2p,eth): remove per-peer dual-connection mechanism, fix #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: #2359

Author: gzliudan

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 {

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

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)
+	}
+}

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
 }
 

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

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