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

Concurrent removePeer callers (BFT broadcast loops, DAO fork timers and
normal teardown) could all pass the Peer(id) lookup before the first one
unregistered the peer, re-running the unregister sequence and flooding the
logs with hundreds of "peer is not registered" warnings on every peer drop.

Mark peer removal with an atomic flag so only the first caller proceeds,
and make downloader.UnregisterPeer treat an already-unregistered peer as a
silent no-op.

Author: gzliudan

eth/downloader/downloader.go

@@ -354,15 +354,44 @@ 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) in the downloader returns
+// errNotRegistered — the exported contract is preserved. That case is logged at
+// Debug level since it is an expected condition, not an anomaly: the peer's
+// RegisterPeer call may have failed or been rejected, or its removal may have
+// raced ahead of registration (see ProtocolManager.removePeer and
+// registerDownloaderPeer). The caller is responsible for tolerating it as an
+// idempotent no-op.
+//
+// The caller must guarantee at-most-once semantics per peer. Today only
+// ProtocolManager.removePeer calls this, and its markRemoved guard ensures a
+// peer is unregistered at most once. errNotRegistered here can mean the peer
+// was never registered in the downloader (its RegisterPeer call failed or was
+// rejected), but it can also mean the removal raced ahead of the downloader
+// registration: removePeer ran in the window between pm.peers.Register and
+// pm.downloader.RegisterPeer in handle. handle closes that gap by re-checking
+// the peer's removed flag after RegisterPeer and undoing the registration, so
+// no stale downloader entry is left behind. If a second caller is ever added,
+// it must uphold the same at-most-once guarantee.
 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: the peer was never registered in the downloader (its
+			// RegisterPeer call failed or was rejected), or its removal raced
+			// ahead of registration. Keep it quiet — this is not an anomaly
+			// worth a warning. The error is still returned so the exported
+			// contract is preserved; the sole caller (removePeer) treats an
+			// already-unregistered peer as an idempotent no-op.
+			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,37 @@ func TestRequestTTL(t *testing.T) {
 		t.Fatalf("ttlLimit (%v) is below rttMaxEstimate (%v)", ttlLimit, rttMaxEstimate)
 	}
 }
+
+// TestDownloaderUnregisterPeerTwice verifies that unregistering a peer that is
+// no longer present returns errNotRegistered, preserving the exported contract.
+// Duplicate removals are normally prevented upstream by
+// ProtocolManager.removePeer's markRemoved guard; when one slips through,
+// removePeer tolerates errNotRegistered as an idempotent no-op.
+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 in the downloader returns errNotRegistered.
+// That path is expected for peers whose RegisterPeer call was rejected, so it
+// is logged at debug level rather than as a warning.
+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,53 @@ func (pm *ProtocolManager) removePeer(id string) {
 	if peer == nil {
 		return
 	}
+	// Claim the removal exactly once. Concurrent callers (BFT broadcast loops,
+	// DAO fork timers, normal teardown) that all passed the lookup above would
+	// otherwise re-run the unregister sequence and flood the logs with
+	// duplicate "peer is not registered" warnings.
+	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 is expected to succeed here: only removePeer removes peers
+	// from the set, and the markRemoved guard above means this peer was still
+	// present when the removal was claimed. If it ever failed, the peer would
+	// stay in the set with `removed` already set, leaving it unreachable for
+	// future removals — so treat a failure as a genuine anomaly, not a
+	// retryable condition.
 	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 window between
+// pm.peers.Register and this call. A BFT broadcaster (BroadcastVote /
+// BroadcastTimeout / BroadcastSyncInfo) can remove a peer in that gap: its
+// removePeer claims the removal, but the downloader UnregisterPeer there hits
+// errNotRegistered (the peer is not yet registered in the downloader), so
+// without this recheck the downloader would keep a stale entry that blocks a
+// reconnect of the same node id (errAlreadyRegistered). Returns DiscUselessPeer
+// to abort the handshake when the removal was already claimed, matching the
+// disconnect reason removePeer uses.
+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() {
+		pm.downloader.UnregisterPeer(p.id)
+		return p2p.DiscUselessPeer
+	}
+	return nil
+}
+
 func (pm *ProtocolManager) Start(maxPeers int) {
 	pm.maxPeers = maxPeers
 
@@ -390,7 +424,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,138 @@ 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 removePeer is a no-op
+// once the peer is gone, so duplicate removals (BFT broadcast loops, DAO fork
+// timers and normal teardown) cannot re-run the unregister sequence.
+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)
+	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 this second call is
+	// already short-circuited by the peer == nil lookup above (the first
+	// removal took the peer out of the set), so it does NOT exercise
+	// markRemoved's atomic branch. That 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 (BFT broadcast loops, DAO fork timers and normal
+// teardown racing each other) remove the peer 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)
+	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())
+	}
+}
+
+// TestRegisterDownloaderPeerUndoesRacedRemoval reproduces the window in handle()
+// between pm.peers.Register and the downloader registration, where a BFT
+// broadcaster (BroadcastVote / BroadcastTimeout / BroadcastSyncInfo) can remove
+// the peer. removePeer claims the removal, but its downloader.UnregisterPeer
+// hits errNotRegistered (the peer is not yet registered in the downloader), so
+// without the recheck in registerDownloaderPeer the downloader would keep a
+// stale entry that blocks a reconnect of the same node id (errAlreadyRegistered).
+func TestRegisterDownloaderPeerUndoesRacedRemoval(t *testing.T) {
+	pm, _ := newTestProtocolManagerMust(t, downloader.FullSync, 0, nil, nil)
+	defer pm.Stop()
+
+	// Build a real peer and register it in pm.peers only — the same state a
+	// peer is in while handle() is between pm.peers.Register and the downloader
+	// registration.
+	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)
+	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")
+	}
+	// handle() now completes the downloader registration; the recheck must
+	// detect the claimed removal, undo the registration and abort the handshake
+	// with the same disconnect reason removePeer uses.
+	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)
+	}
+}

eth/peer.go

@@ -21,6 +21,7 @@ import (
 	"fmt"
 	"math/big"
 	"sync"
+	"sync/atomic"
 	"time"
 
 	"github.com/XinFinOrg/XDPoSChain/common"
@@ -109,6 +110,13 @@ type peer struct {
 
 	term chan struct{} // Termination channel to stop the broadcaster
 
+	// removed is set exactly once when the peer is removed from the peer set.
+	// It makes peer removal idempotent: concurrent removePeer calls (from the
+	// BFT broadcast loops, DAO fork timers and normal teardown) that all pass
+	// the Peer(id) lookup race can no longer re-run the unregister sequence
+	// and flood the logs with duplicate "peer is not registered" warnings.
+	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 +293,13 @@ func (p *peer) announceTransactions() {
 	}
 }
 
+// markRemoved claims the peer's removal. It returns true only for the first
+// caller; concurrent removePeer calls all observe the flag already set and
+// become no-ops, 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,34 @@ 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, so concurrent removePeer calls cannot re-run the unregister sequence.
+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 the contract removePeer relies on:
+// unregistering an already-removed peer reports errNotRegistered, which the
+// idempotency guard in removePeer prevents from ever being hit.
+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)
+	}
+}