diff --git a/eth/backend.go b/eth/backend.go index 05b9fa7adf..5b9721ad60 100644 --- a/eth/backend.go +++ b/eth/backend.go @@ -479,7 +479,7 @@ func (s *Ethereum) Start() error { s.handler.Start(s.p2pServer.MaxPeers) // Start the connection manager with inclusion-based peer protection. - s.dropper.Start(s.p2pServer, func() bool { return !s.Synced() }, s.handler.peerStats.GetAllPeerStats) + s.dropper.Start(s.p2pServer, func() bool { return !s.Synced() }, s.handler.peerStats.GetAllPeerStats, s.handler.peerStats.Prune) // Subscribe to chain events for the filterMaps head updater. s.fmHeadSub = s.blockchain.SubscribeChainEvent(s.fmHeadEventCh) diff --git a/eth/dropper.go b/eth/dropper.go index 0f80e1e5b0..b15d801bde 100644 --- a/eth/dropper.go +++ b/eth/dropper.go @@ -122,7 +122,8 @@ type dropper struct { maxInboundPeers int // maximum number of inbound peers peersFunc getPeersFunc syncingFunc getSyncingFunc - peerStatsFunc getPeerStatsFunc // optional: inclusion stats for protection + peerStatsFunc getPeerStatsFunc // optional: inclusion stats for protection + pruneStatsFunc func(map[string]bool) // optional: reclaim stats for disconnected peers // peerDropTimer introduces churn if we are close to limit capacity. // We handle Dialed and Inbound connections separately @@ -152,12 +153,13 @@ func newDropper(maxDialPeers, maxInboundPeers int) *dropper { return cm } -// Start the dropper. peerStatsFunc is optional (nil disables inclusion -// protection). -func (cm *dropper) Start(srv *p2p.Server, syncingFunc getSyncingFunc, peerStatsFunc getPeerStatsFunc) { +// Start the dropper. peerStatsFunc and pruneStatsFunc are optional (nil +// disables inclusion protection and stats pruning respectively). +func (cm *dropper) Start(srv *p2p.Server, syncingFunc getSyncingFunc, peerStatsFunc getPeerStatsFunc, pruneStatsFunc func(map[string]bool)) { cm.peersFunc = srv.Peers cm.syncingFunc = syncingFunc cm.peerStatsFunc = peerStatsFunc + cm.pruneStatsFunc = pruneStatsFunc cm.wg.Add(1) go cm.loop() } @@ -217,6 +219,21 @@ func (cm *dropper) dropRandomPeer() bool { return true } +// pruneStats reclaims stats for peers that are no longer connected. It builds +// the currently-connected id set and hands it to the stats pruner. No-op when +// pruning is disabled (nil pruneStatsFunc). +func (cm *dropper) pruneStats() { + if cm.pruneStatsFunc == nil { + return + } + peers := cm.peersFunc() + keep := make(map[string]bool, len(peers)) + for _, p := range peers { + keep[p.ID().String()] = true + } + cm.pruneStatsFunc(keep) +} + // protectedPeers computes the set of peers that should not be dropped based // on inclusion stats. Each protection category independently selects its // top-N peers per inbound/dialed pool; the union is returned. @@ -295,6 +312,11 @@ func (cm *dropper) loop() { for { select { case <-cm.peerDropTimer.C: + // Reclaim stats entries for peers that are no longer connected, + // covering the rare orphan left when a peer signal races with its + // NotifyPeerDrop. Done every tick (independent of syncing) since + // disconnects happen during sync too. + cm.pruneStats() // Drop a random peer if we are not syncing and the peer count is close to the limit. if !cm.syncingFunc() { cm.dropRandomPeer() diff --git a/eth/peerstats/peerstats.go b/eth/peerstats/peerstats.go index 146226a187..4fa3e49fff 100644 --- a/eth/peerstats/peerstats.go +++ b/eth/peerstats/peerstats.go @@ -162,16 +162,35 @@ func (s *Stats) NotifyRequestResult(peer string, latency time.Duration, timeout ps.lastLatencySample = time.Now() } -// NotifyPeerDrop removes a peer's stats on disconnect. A rare stale -// latency sample racing with the drop may recreate the peer entry with -// one sample; that entry can never earn protection (MinLatencySamples -// guard) and is harmless. +// NotifyPeerDrop removes a peer's stats on disconnect. +// +// A signal (NotifyRequestResult or NotifyBlock) for the same peer can race +// with the drop and land just after this deletion, recreating an orphan +// entry that no future NotifyPeerDrop will ever clean. Such orphans are +// never read — the dropper only looks up currently-connected peers — but +// left alone they accumulate for the node's lifetime. Prune reclaims them. func (s *Stats) NotifyPeerDrop(peer string) { s.mu.Lock() defer s.mu.Unlock() delete(s.peers, peer) } +// Prune removes stats for every peer not present in keep. The dropper calls +// this periodically with the set of currently-connected peer IDs to reclaim +// orphan entries left by a signal that raced with NotifyPeerDrop (see there). +// Pruning a still-connected peer that only just gained an entry is harmless: +// it resets a handful of early samples that self-heal on the peer's next +// activity, and such a peer cannot yet meet the protection thresholds. +func (s *Stats) Prune(keep map[string]bool) { + s.mu.Lock() + defer s.mu.Unlock() + for id := range s.peers { + if !keep[id] { + delete(s.peers, id) + } + } +} + // GetAllPeerStats returns a snapshot of per-peer stats. Called by the // dropper every few minutes; allocation cost is negligible at that rate. func (s *Stats) GetAllPeerStats() map[string]PeerStats { diff --git a/eth/peerstats/peerstats_test.go b/eth/peerstats/peerstats_test.go index 3b242eac15..4a46155a9d 100644 --- a/eth/peerstats/peerstats_test.go +++ b/eth/peerstats/peerstats_test.go @@ -183,6 +183,38 @@ func TestNotifyPeerDropClearsStats(t *testing.T) { } } +// TestPruneRemovesDisconnectedPeers verifies Prune drops entries for peers +// absent from the keep set (e.g. an orphan recreated by a signal that raced +// with NotifyPeerDrop) while retaining still-connected peers. +func TestPruneRemovesDisconnectedPeers(t *testing.T) { + s := New() + s.NotifyRequestResult("connected", 100*time.Millisecond, false) + s.NotifyRequestResult("orphan", 100*time.Millisecond, false) + + s.Prune(map[string]bool{"connected": true}) + + stats := s.GetAllPeerStats() + if _, ok := stats["orphan"]; ok { + t.Fatal("orphan stats should be pruned when not in keep set") + } + if _, ok := stats["connected"]; !ok { + t.Fatal("connected peer stats should survive pruning") + } +} + +// TestPruneEmptyKeepClearsAll verifies an empty keep set removes every entry. +func TestPruneEmptyKeepClearsAll(t *testing.T) { + s := New() + s.NotifyRequestResult("peerA", 100*time.Millisecond, false) + s.NotifyRequestResult("peerB", 100*time.Millisecond, false) + + s.Prune(map[string]bool{}) + + if n := len(s.GetAllPeerStats()); n != 0 { + t.Fatalf("expected all entries pruned, got %d", n) + } +} + // TestStaleRequestLatencyAfterDrop documents the accepted behavior: a // late sample after NotifyPeerDrop recreates a 1-sample entry. The // dropper's MinLatencySamples=100 guard ensures this is harmless.