fix(eth): make peer removal idempotent to stop unregister flood

eth/downloader/downloader.go

@@ -354,15 +354,24 @@ func (d *Downloader) RegisterLightPeer(id string, version int, peer LightPeer) e
 	return d.RegisterPeer(id, version, &lightPeerWrapper{peer})
 }
 
-// UnregisterPeer remove a peer from the known list, preventing any action from
+// UnregisterPeer removes a peer from the known list, preventing any action from
 // the specified peer. An effort is also made to return any pending fetches into
 // the queue.
+//
+// Unregistering a peer that is not (or no longer) registered returns
+// errNotRegistered without side effects, so repeated or racing calls are safe:
+// the cleanup (queue revocation and peer drop event) runs at most once.
 func (d *Downloader) UnregisterPeer(id string) error {
 	// Unregister the peer from the active peer set and revoke any fetch tasks
 	logger := log.New("peer", id)
 	logger.Trace("Unregistering sync peer")
 	if err := d.peers.Unregister(id); err != nil {
-		logger.Warn("Failed to unregister sync peer", "err", err)
+		if errors.Is(err, errNotRegistered) {
+			// Expected: never registered, or removal raced ahead of registration.
+			logger.Debug("Sync peer was never registered")
+		} else {
+			logger.Warn("Failed to unregister sync peer", "err", err)
+		}
 		return err
 	}
 	d.queue.Revoke(id)

eth/downloader/downloader_test.go

@@ -2299,3 +2299,32 @@ func TestRequestTTL(t *testing.T) {
 		t.Fatalf("ttlLimit (%v) is below rttMaxEstimate (%v)", ttlLimit, rttMaxEstimate)
 	}
 }
+
+// TestDownloaderUnregisterPeerTwice verifies that a second unregister of an
+// already-removed peer returns errNotRegistered, per the exported contract.
+func TestDownloaderUnregisterPeerTwice(t *testing.T) {
+	dl := newTester()
+	defer dl.terminate()
+
+	chain := newTestChain(1, testGenesis)
+	if err := dl.newPeer("unreg", 62, chain); err != nil {
+		t.Fatalf("failed to register test peer: %v", err)
+	}
+	if err := dl.downloader.UnregisterPeer("unreg"); err != nil {
+		t.Fatalf("first unregister failed: %v", err)
+	}
+	if err := dl.downloader.UnregisterPeer("unreg"); err != errNotRegistered {
+		t.Fatalf("second unregister error mismatch: got %v want %v", err, errNotRegistered)
+	}
+}
+
+// TestDownloaderUnregisterPeerNeverRegistered verifies that unregistering a
+// peer that was never registered returns errNotRegistered.
+func TestDownloaderUnregisterPeerNeverRegistered(t *testing.T) {
+	dl := newTester()
+	defer dl.terminate()
+
+	if err := dl.downloader.UnregisterPeer("ghost"); err != errNotRegistered {
+		t.Fatalf("unregistering a never-registered peer error mismatch: got %v want %v", err, errNotRegistered)
+	}
+}

eth/handler.go

@@ -281,19 +281,42 @@ func (pm *ProtocolManager) removePeer(id string) {
 	if peer == nil {
 		return
 	}
+	// Claim the removal exactly once; concurrent callers become no-ops.
+	if !peer.markRemoved() {
+		return
+	}
 	log.Debug("Removing Ethereum peer", "peer", id)
 
 	// Unregister the peer from the downloader and Ethereum peer set
 	pm.downloader.UnregisterPeer(id)
 	pm.txFetcher.Drop(id)
 
+	// Unregister should succeed: the guard above guarantees the peer is still
+	// in the set.
 	if err := pm.peers.Unregister(id); err != nil {
 		log.Debug("Peer removal failed", "peer", id, "err", err)
 	}
 	// Hard disconnect at the networking layer
 	peer.Peer.Disconnect(p2p.DiscUselessPeer)
 }
 
+// registerDownloaderPeer registers the peer with the downloader, undoing the
+// registration if the peer's removal was claimed in the meantime; otherwise a
+// stale entry would block a reconnect of the same node id. Returns
+// DiscUselessPeer to abort the handshake, matching removePeer's disconnect
+// reason.
+func (pm *ProtocolManager) registerDownloaderPeer(p *peer) error {
+	if err := pm.downloader.RegisterPeer(p.id, p.version, p); err != nil {
+		return err
+	}
+	if p.removed.Load() {
+		// Undo the registration; UnregisterPeer is a no-op if already removed.
+		pm.downloader.UnregisterPeer(p.id)
+		return p2p.DiscUselessPeer
+	}
+	return nil
+}
+
 func (pm *ProtocolManager) Start(maxPeers int) {
 	pm.maxPeers = maxPeers
 
@@ -390,7 +413,7 @@ func (pm *ProtocolManager) handle(p *peer) error {
 	defer pm.removePeer(p.id)
 
 	// 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 {
+	if err := pm.registerDownloaderPeer(p); err != nil {
 		return err
 	}
 	p.Log().Info("Register peer", "nodeid", p.ID().String(), "version", p.version, "addr", p.RemoteAddr())

eth/handler_test.go

@@ -21,6 +21,7 @@ import (
 	"math"
 	"math/big"
 	"math/rand"
+	"sync"
 	"testing"
 	"time"
 
@@ -36,6 +37,7 @@ import (
 	"github.com/XinFinOrg/XDPoSChain/eth/ethconfig"
 	"github.com/XinFinOrg/XDPoSChain/event"
 	"github.com/XinFinOrg/XDPoSChain/p2p"
+	"github.com/XinFinOrg/XDPoSChain/p2p/enode"
 	"github.com/XinFinOrg/XDPoSChain/params"
 )
 
@@ -765,3 +767,153 @@ func daoChallengeChainConfig(daoForkSupport bool) *params.ChainConfig {
 
 	return config
 }
+
+// waitForPeerRegistration blocks until the peer with the given id has been
+// registered by the protocol manager's handle goroutine.
+func waitForPeerRegistration(t *testing.T, pm *ProtocolManager, id string) {
+	t.Helper()
+	deadline := time.After(2 * time.Second)
+	for pm.peers.Peer(id) == nil {
+		select {
+		case <-deadline:
+			t.Fatalf("test peer %s was not registered in time", id)
+		case <-time.After(10 * time.Millisecond):
+		}
+	}
+}
+
+// TestProtocolManagerRemovePeerIdempotent verifies that removing an already
+// removed peer is a silent no-op.
+func TestProtocolManagerRemovePeerIdempotent(t *testing.T) {
+	pm, _ := newTestProtocolManagerMust(t, downloader.FullSync, 0, nil, nil)
+	defer pm.Stop()
+
+	// Register a peer through the normal protocol handshake path.
+	tp, errc := newTestPeer("test-peer", xdc165, pm, true)
+	// Stop the broadcast goroutines started by peers.Register; nothing in the
+	// production teardown path closes the peer's term channel.
+	defer tp.peer.close()
+	defer tp.close()
+	defer tp.app.Close()
+	defer func() {
+		select {
+		case <-errc:
+		default:
+		}
+	}()
+	waitForPeerRegistration(t, pm, tp.id)
+
+	if pm.peers.Len() != 1 {
+		t.Fatalf("peer set size mismatch: got %d want 1", pm.peers.Len())
+	}
+	// The first removal performs the full unregister sequence.
+	pm.removePeer(tp.id)
+	if pm.peers.Peer(tp.id) != nil {
+		t.Fatal("peer still registered after first removePeer")
+	}
+	if pm.peers.Len() != 0 {
+		t.Fatalf("peer set size mismatch after removal: got %d want 0", pm.peers.Len())
+	}
+	// A duplicate removal must be a silent no-op. Note it is short-circuited by
+	// the peer == nil lookup above, so markRemoved's atomic branch is covered
+	// by TestPeerMarkRemovedOnce and TestProtocolManagerRemovePeerConcurrent.
+	pm.removePeer(tp.id)
+	if pm.peers.Len() != 0 {
+		t.Fatalf("peer set size mismatch after second removePeer: got %d want 0", pm.peers.Len())
+	}
+}
+
+// TestProtocolManagerRemovePeerConcurrent verifies that concurrent removePeer
+// calls for the same peer remove it exactly once, without panicking or racing
+// on the removal flag.
+func TestProtocolManagerRemovePeerConcurrent(t *testing.T) {
+	pm, _ := newTestProtocolManagerMust(t, downloader.FullSync, 0, nil, nil)
+	defer pm.Stop()
+
+	// Register a peer through the normal protocol handshake path.
+	tp, errc := newTestPeer("test-peer", xdc165, pm, true)
+	// Stop the broadcast goroutines started by peers.Register; nothing in the
+	// production teardown path closes the peer's term channel.
+	defer tp.peer.close()
+	defer tp.close()
+	defer tp.app.Close()
+	defer func() {
+		select {
+		case <-errc:
+		default:
+		}
+	}()
+	waitForPeerRegistration(t, pm, tp.id)
+
+	var wg sync.WaitGroup
+	for i := 0; i < 16; i++ {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			pm.removePeer(tp.id)
+		}()
+	}
+	wg.Wait()
+
+	if pm.peers.Peer(tp.id) != nil {
+		t.Fatal("peer still registered after concurrent removePeer calls")
+	}
+	if pm.peers.Len() != 0 {
+		t.Fatalf("peer set size mismatch after concurrent removal: got %d want 0", pm.peers.Len())
+	}
+	// The downloader must not retain a stale entry that blocks re-registration.
+	// Poll briefly in case handle() is still undoing its registration.
+	deadline := time.After(2 * time.Second)
+	for {
+		if err := pm.downloader.RegisterPeer(tp.id, tp.version, tp); err == nil {
+			break
+		}
+		select {
+		case <-deadline:
+			t.Fatal("stale downloader entry blocks re-registration")
+		case <-time.After(10 * time.Millisecond):
+		}
+	}
+	// Undo the test's own registration.
+	pm.downloader.UnregisterPeer(tp.id)
+}
+
+// TestRegisterDownloaderPeerUndoesRacedRemoval reproduces the window in handle()
+// between pm.peers.Register and the downloader registration, where a BFT
+// broadcaster can remove the peer. Without the recheck in
+// registerDownloaderPeer, the downloader would keep a stale entry that blocks
+// a reconnect of the same node id.
+func TestRegisterDownloaderPeerUndoesRacedRemoval(t *testing.T) {
+	pm, _ := newTestProtocolManagerMust(t, downloader.FullSync, 0, nil, nil)
+	defer pm.Stop()
+
+	// Register the peer in pm.peers only — the state handle() is in mid-window.
+	app, net := p2p.MsgPipe()
+	defer app.Close()
+	var id enode.ID
+	rand.Read(id[:])
+	p := pm.newPeer(xdc165, p2p.NewPeer(id, "race-peer", nil), net, pm.txpool.Get)
+	// peers.Register starts the peer's broadcast goroutines; close the term
+	// channel so they terminate when the test ends.
+	defer p.close()
+	if err := pm.peers.Register(p); err != nil {
+		t.Fatalf("failed to register test peer: %v", err)
+	}
+	if pm.peers.Len() != 1 {
+		t.Fatalf("peer set size mismatch: got %d want 1", pm.peers.Len())
+	}
+	// A BFT broadcaster's failing send removes the peer inside the window.
+	pm.removePeer(p.id)
+	if pm.peers.Peer(p.id) != nil {
+		t.Fatal("peer still present after removePeer")
+	}
+	// The recheck must undo the registration and abort the handshake.
+	if err := pm.registerDownloaderPeer(p); err != p2p.DiscUselessPeer {
+		t.Fatalf("registerDownloaderPeer should abort with DiscUselessPeer a handshake whose removal was already claimed, got: %v", err)
+	}
+	// A reconnect of the same node id must not be blocked by a stale entry.
+	if err := pm.downloader.RegisterPeer(p.id, p.version, p); err != nil {
+		t.Fatalf("reconnect blocked by stale downloader entry: %v", err)
+	}
+	pm.downloader.UnregisterPeer(p.id)
+}

eth/peer.go

@@ -21,6 +21,7 @@ import (
 	"fmt"
 	"math/big"
 	"sync"
+	"sync/atomic"
 	"time"
 
 	"github.com/XinFinOrg/XDPoSChain/common"
@@ -109,6 +110,9 @@ type peer struct {
 
 	term chan struct{} // Termination channel to stop the broadcaster
 
+	// removed is set exactly once to make peer removal idempotent.
+	removed atomic.Bool
+
 	knownVote     mapset.Set[common.Hash] // Set of BFT Vote known to be known by this peer
 	knownTimeout  mapset.Set[common.Hash] // Set of BFT timeout known to be known by this peer
 	knownSyncInfo mapset.Set[common.Hash] // Set of BFT Sync Info known to be known by this peer
@@ -285,6 +289,12 @@ func (p *peer) announceTransactions() {
 	}
 }
 
+// markRemoved claims the peer's removal, returning true only for the first
+// caller so the unregister sequence runs exactly once per peer.
+func (p *peer) markRemoved() bool {
+	return !p.removed.Swap(true)
+}
+
 // close signals the broadcast goroutine to terminate.
 func (p *peer) close() {
 	close(p.term)

eth/peer_test.go

@@ -20,3 +20,32 @@ func TestPeerSetRegisterRejectsDuplicateID(t *testing.T) {
 		t.Fatalf("registered peer replaced: got %p want %p", got, first)
 	}
 }
+
+// TestPeerMarkRemovedOnce verifies that a peer's removal is claimed exactly once.
+func TestPeerMarkRemovedOnce(t *testing.T) {
+	p := &peer{id: "once"}
+	if !p.markRemoved() {
+		t.Fatal("first markRemoved should claim the removal")
+	}
+	for i := 0; i < 10; i++ {
+		if p.markRemoved() {
+			t.Fatalf("markRemoved should not claim a removal after it was already claimed (iteration %d)", i)
+		}
+	}
+}
+
+// TestPeerSetUnregisterTwice documents that unregistering an already-removed
+// peer reports errNotRegistered.
+func TestPeerSetUnregisterTwice(t *testing.T) {
+	peers := newPeerSet()
+	p := &peer{id: "twice"}
+	if err := peers.Register(p); err != nil {
+		t.Fatalf("register failed: %v", err)
+	}
+	if err := peers.Unregister("twice"); err != nil {
+		t.Fatalf("first unregister failed: %v", err)
+	}
+	if err := peers.Unregister("twice"); err != errNotRegistered {
+		t.Fatalf("second unregister error mismatch: got %v want %v", err, errNotRegistered)
+	}
+}