From fccabb26cbd08a8c7ad0bd97486353061437fcf2 Mon Sep 17 00:00:00 2001 From: Daniel Liu <139250065@qq.com> Date: Tue, 18 Aug 2026 06:06:04 +0800 Subject: [PATCH] fix(eth): make peer removal idempotent to stop unregister flood Concurrent removePeer callers (BFT broadcast loops, DAO fork timers, normal teardown, and the downloader/fetcher drop callbacks) can all pass the Peer(id) lookup before the first caller unregisters the peer, so each re-runs the unregister sequence and floods the logs with "peer is not registered" warnings on every peer drop. Mark peer removal with an atomic flag so the unregister sequence runs exactly once, and let downloader.UnregisterPeer treat an unregistered peer as a silent no-op. In the same race window, handle() could register a peer in the downloader after its removal was claimed, leaving a stale entry that blocks a reconnect of the same node id; re-check the flag after registering and undo the registration. --- eth/downloader/downloader.go | 13 ++- eth/downloader/downloader_test.go | 29 ++++++ eth/handler.go | 25 ++++- eth/handler_test.go | 152 ++++++++++++++++++++++++++++++ eth/peer.go | 10 ++ eth/peer_test.go | 29 ++++++ 6 files changed, 255 insertions(+), 3 deletions(-) diff --git a/eth/downloader/downloader.go b/eth/downloader/downloader.go index 1ae7f9d8c57f..c05e92f8813f 100644 --- a/eth/downloader/downloader.go +++ b/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) diff --git a/eth/downloader/downloader_test.go b/eth/downloader/downloader_test.go index c872c27ac393..f7195caf0dbb 100644 --- a/eth/downloader/downloader_test.go +++ b/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) + } +} diff --git a/eth/handler.go b/eth/handler.go index 1d2b63c17931..168ec3afc3e9 100644 --- a/eth/handler.go +++ b/eth/handler.go @@ -281,12 +281,18 @@ 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) } @@ -294,6 +300,23 @@ func (pm *ProtocolManager) removePeer(id string) { 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()) diff --git a/eth/handler_test.go b/eth/handler_test.go index f682599b96a3..d017b3bc5838 100644 --- a/eth/handler_test.go +++ b/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) +} diff --git a/eth/peer.go b/eth/peer.go index daf97d0b23e0..7e99874c0c15 100644 --- a/eth/peer.go +++ b/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) diff --git a/eth/peer_test.go b/eth/peer_test.go index 95c43062db15..adf135615c0a 100644 --- a/eth/peer_test.go +++ b/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) + } +}