eth/peerstats: prune stats for disconnected peers

A NotifyRequestResult or NotifyBlock for a peer can race with its
NotifyPeerDrop and land just after the deletion, recreating an orphan
entry that no later NotifyPeerDrop will ever clean. The dropper only
reads stats for currently-connected peers, so such orphans are never
used, but they accumulate for the node's lifetime.

Add Stats.Prune(keep) and call it from the dropper loop each tick with
the currently-connected peer id set (independent of syncing, since peers
disconnect during sync too), reclaiming any orphaned entries.
This commit is contained in:
Csaba Kiraly 2026-07-15 13:08:29 +02:00
parent 7b4732c197
commit 0eec801495
4 changed files with 82 additions and 9 deletions

View file

@ -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)

View file

@ -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()

View file

@ -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 {

View file

@ -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.