diff --git a/eth/downloader/downloader.go b/eth/downloader/downloader.go index 1ae7f9d8c57..c05e92f8813 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 c872c27ac39..f7195caf0db 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 1d2b63c1793..168ec3afc3e 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 f682599b96a..d017b3bc583 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 daf97d0b23e..7e99874c0c1 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 95c43062db1..adf135615c0 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) + } +}