From 91fed6db6bd275b43b96e6093eb69b534eaa2e34 Mon Sep 17 00:00:00 2001 From: Csaba Kiraly Date: Wed, 15 Jul 2026 15:40:59 +0200 Subject: [PATCH] eth/peerstats: replace sample-count and staleness gates with activity EMA MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Latency-protection eligibility used two mechanisms with an exploitable seam between them: a cumulative sample count (front-loadable — 100 samples in one burst qualified a peer for its connection lifetime) and a last-sample timestamp (maintainable with a single sample per 10-minute window). Together the ongoing cost of holding eligibility was ~6 samples per hour after an initial burst. Replace both with a single block-decayed EMA of accepted-delivery samples (LatencyActivity, ~10-minute half-life, folded and decayed in NotifyBlock like the inclusion EMAs). The gate MinLatencyActivity = 0.2 demands roughly one accepted fetch per minute, sustained: eligibility now expires on its own and cannot be front-loaded. Timeouts update the EMA and counters but never feed activity, so a peer cannot become eligible by timing out. Once a formerly-active peer's activity fully decays (~75 minutes of silence), its latency state is forgotten entirely — a frozen fast EMA cannot be re-armed later by rebuilding activity alone. The reset is gated on success history so timeout-only peers keep their penalty record. The dropper keeps the raw 1/EMA ranking: among peers doing sustained useful work, the lowest-latency ones win protection. --- eth/dropper.go | 21 +++---- eth/dropper_test.go | 55 ++++++++--------- eth/peerstats/peerstats.go | 77 +++++++++++++++++------- eth/peerstats/peerstats_test.go | 103 +++++++++++++++++++++++++------- 4 files changed, 169 insertions(+), 87 deletions(-) diff --git a/eth/dropper.go b/eth/dropper.go index b15d801bde..53b425e3ac 100644 --- a/eth/dropper.go +++ b/eth/dropper.go @@ -75,19 +75,14 @@ var protectionCategories = []protectionCategory{ {func(s peerstats.PeerStats) float64 { return s.RecentFinalized }, inclusionProtectionFrac}, // Recent finalized {func(s peerstats.PeerStats) float64 { return s.RecentIncluded }, inclusionProtectionFrac}, // Recent included {func(s peerstats.PeerStats) float64 { // Request latency - // Low-latency peers should rank higher. Peers with too few samples - // score 0 so the existing `score > 0` filter excludes them — this - // prevents a single lucky-fast reply from winning protection. Peers - // whose EMA reaches the timeout also score low by this path because - // the reciprocal of a very large duration is tiny but positive; the - // per-pool top-N will still push faster peers ahead of them. - if s.RequestSuccesses+s.RequestTimeouts < peerstats.MinLatencySamples { - return 0 - } - // Freshness gate: a peer that earned a fast EMA but then went - // silent on announcements (no requests → no fresh samples) must - // not keep that score indefinitely. Ignore stale data. - if time.Since(s.LastLatencySample) > peerstats.MaxLatencyStaleness { + // Low-latency peers rank higher. Eligibility requires a sustained + // rate of accepted-delivery samples (block-decayed EMA): scoring 0 + // here lets the `score > 0` filter exclude peers whose EMA rests on + // too little, too old, or front-loaded evidence — eligibility + // expires on its own when the useful work stops. Peers whose EMA + // approaches the fetch timeout score tiny-but-positive via the + // reciprocal; per-pool top-N pushes faster peers ahead of them. + if s.LatencyActivity < peerstats.MinLatencyActivity { return 0 } if s.RequestLatencyEMA <= 0 { diff --git a/eth/dropper_test.go b/eth/dropper_test.go index 67f24a064d..92556f4250 100644 --- a/eth/dropper_test.go +++ b/eth/dropper_test.go @@ -243,18 +243,15 @@ func TestProtectedByPoolRequestLatencyBasic(t *testing.T) { // Three peers have enough samples; the two fastest should win. stats[dialed[0].ID().String()] = peerstats.PeerStats{ RequestLatencyEMA: 50 * time.Millisecond, - RequestSuccesses: peerstats.MinLatencySamples, - LastLatencySample: time.Now(), + LatencyActivity: peerstats.MinLatencyActivity, } stats[dialed[1].ID().String()] = peerstats.PeerStats{ RequestLatencyEMA: 100 * time.Millisecond, - RequestSuccesses: peerstats.MinLatencySamples, - LastLatencySample: time.Now(), + LatencyActivity: peerstats.MinLatencyActivity, } stats[dialed[2].ID().String()] = peerstats.PeerStats{ RequestLatencyEMA: 2 * time.Second, - RequestSuccesses: peerstats.MinLatencySamples, - LastLatencySample: time.Now(), + LatencyActivity: peerstats.MinLatencyActivity, } protected := protectedPeersByPool(nil, dialed, stats) @@ -273,22 +270,23 @@ func TestProtectedByPoolRequestLatencyBasic(t *testing.T) { } } -// TestProtectedByPoolRequestLatencyBootstrapGuard verifies that peers with -// fewer than MinLatencySamples do not earn latency-based protection, even -// if their few samples indicate very low latency. +// TestProtectedByPoolRequestLatencyBootstrapGuard verifies that peers whose +// accepted-delivery activity rate is below MinLatencyActivity do not earn +// latency-based protection, even if their few samples indicate very low +// latency. func TestProtectedByPoolRequestLatencyBootstrapGuard(t *testing.T) { dialed := makePeers(20) stats := make(map[string]peerstats.PeerStats) - // A lucky-fast peer with only 1 sample — must NOT be protected. + // A lucky-fast peer without sustained activity — must NOT be protected. stats[dialed[0].ID().String()] = peerstats.PeerStats{ RequestLatencyEMA: 1 * time.Millisecond, RequestSuccesses: 1, + LatencyActivity: peerstats.MinLatencyActivity / 2, } // A warmed-up but slower peer — should be protected on latency. stats[dialed[1].ID().String()] = peerstats.PeerStats{ RequestLatencyEMA: 500 * time.Millisecond, - RequestSuccesses: peerstats.MinLatencySamples, - LastLatencySample: time.Now(), + LatencyActivity: peerstats.MinLatencyActivity, } protected := protectedPeersByPool(nil, dialed, stats) @@ -317,8 +315,7 @@ func TestProtectedByPoolRequestLatencyPerPool(t *testing.T) { for _, p := range inbound { stats[p.ID().String()] = peerstats.PeerStats{ RequestLatencyEMA: 50 * time.Millisecond, - RequestSuccesses: peerstats.MinLatencySamples, - LastLatencySample: time.Now(), + LatencyActivity: peerstats.MinLatencyActivity, } } // Dialed peers are slower (1s) — globally they would all lose, but @@ -326,8 +323,7 @@ func TestProtectedByPoolRequestLatencyPerPool(t *testing.T) { for _, p := range dialed { stats[p.ID().String()] = peerstats.PeerStats{ RequestLatencyEMA: 1 * time.Second, - RequestSuccesses: peerstats.MinLatencySamples, - LastLatencySample: time.Now(), + LatencyActivity: peerstats.MinLatencyActivity, } } @@ -345,34 +341,33 @@ func TestProtectedByPoolRequestLatencyPerPool(t *testing.T) { } } -// TestProtectedByPoolRequestLatencyStale verifies that the freshness gate -// excludes peers whose latency EMA is valid (meeting the sample count and -// fast value) but whose last sample is older than MaxLatencyStaleness. -// A peer cannot serve a burst of fast replies, go silent on announcements, -// and keep latency-based protection indefinitely. +// TestProtectedByPoolRequestLatencyStale verifies that decayed activity +// excludes peers whose latency EMA is fast but whose accepted-delivery +// rate has since fallen below MinLatencyActivity. A peer cannot serve a +// burst of fast replies, go silent on announcements, and keep +// latency-based protection indefinitely. func TestProtectedByPoolRequestLatencyStale(t *testing.T) { dialed := makePeers(20) stats := make(map[string]peerstats.PeerStats) - // Fresh, fast peer — should be protected. + // Active, fast peer — should be protected. stats[dialed[0].ID().String()] = peerstats.PeerStats{ RequestLatencyEMA: 50 * time.Millisecond, - RequestSuccesses: peerstats.MinLatencySamples, - LastLatencySample: time.Now(), + LatencyActivity: peerstats.MinLatencyActivity, } - // Stale, fast peer — was fast, but hasn't answered in too long. - // Same EMA and sample count as the fresh peer; only staleness differs. + // Formerly active, fast peer — same EMA, but its activity has decayed + // below the eligibility threshold. stats[dialed[1].ID().String()] = peerstats.PeerStats{ RequestLatencyEMA: 50 * time.Millisecond, - RequestSuccesses: peerstats.MinLatencySamples, - LastLatencySample: time.Now().Add(-2 * peerstats.MaxLatencyStaleness), + RequestSuccesses: 100, + LatencyActivity: peerstats.MinLatencyActivity * 0.9, } protected := protectedPeersByPool(nil, dialed, stats) if !protected[dialed[0]] { - t.Error("fresh fast peer must be protected") + t.Error("active fast peer must be protected") } if protected[dialed[1]] { - t.Error("stale peer must NOT keep latency protection despite fast EMA") + t.Error("decayed-activity peer must NOT keep latency protection despite fast EMA") } } diff --git a/eth/peerstats/peerstats.go b/eth/peerstats/peerstats.go index 4fa3e49fff..7ab8422781 100644 --- a/eth/peerstats/peerstats.go +++ b/eth/peerstats/peerstats.go @@ -28,8 +28,9 @@ // (computed under txtracker's own lock, then passed in after release) // - NotifyRequestResult(peer, latency, timeout) — per-request outcomes // from the fetcher; timeouts are reported with the timeout value so -// slow peers contribute to the EMA, and the timeout flag increments -// the per-peer timeout counter +// slow peers contribute to the EMA. Non-timeout results are only +// reported for deliveries with pool-accepted txs, and only those feed +// the activity rate gating latency protection // - NotifyPeerDrop(peer) — called from the handler on disconnect package peerstats @@ -50,17 +51,26 @@ const ( // short bursts shouldn't shift the score, sustained behavior should. // Half-life ≈ ln(0.5)/ln(0.99) ≈ 69 samples. latencyEMAAlpha = 0.01 - // MinLatencySamples is the number of latency samples a peer must accumulate - // before its RequestLatencyEMA is considered meaningful for protection. - // Prevents a single lucky-fast reply from displacing established peers. - MinLatencySamples = 100 - // MaxLatencyStaleness is the oldest allowed age of a peer's last - // latency sample before their RequestLatencyEMA is disregarded for - // protection. Prevents a peer from earning a fast score during a - // burst of activity and then holding protection indefinitely by - // going silent on tx announcements (no further requests → no fresh - // samples → EMA frozen at its last value). - MaxLatencyStaleness = 10 * time.Minute + // EMA smoothing factor for the per-block latency-sample activity rate. + // Half-life ≈ 50 chain heads (~10 minutes on 12s blocks): eligibility + // for latency protection must be continuously maintained at roughly + // this cadence, it cannot be front-loaded in a burst and then held. + latencyActivityAlpha = 0.014 + // MinLatencyActivity is the minimum sustained rate of accepted-delivery + // latency samples (per block, EMA-smoothed) a peer must maintain for its + // RequestLatencyEMA to be considered for protection. 0.2 ≈ one accepted + // fetch per five blocks (~1/minute). Replaces both an absolute sample + // count (front-loadable) and a last-sample staleness check (maintainable + // with one sample per window): a decaying rate expires on its own and + // demands sustained useful work. + MinLatencyActivity = 0.2 + // latencyResetThreshold is the activity level below which a peer's + // latency state is forgotten entirely. Without this, a peer could + // earn a fast EMA, go silent (activity decays, eligibility lost) and + // later re-arm the frozen EMA by rebuilding activity alone. Once + // activity has decayed this far (~75 minutes of silence from the + // eligibility threshold), the peer starts over as a stranger. + latencyResetThreshold = 0.001 ) // PeerStats is the exported per-peer snapshot returned by GetAllPeerStats. @@ -68,9 +78,9 @@ type PeerStats struct { RecentFinalized float64 // EMA of per-block finalization credits (slow) RecentIncluded float64 // EMA of per-block inclusions (fast) RequestLatencyEMA time.Duration // Slow EMA of tx-request response latency (timeouts count as the timeout value) - RequestSuccesses int64 // Requests answered before timeout + RequestSuccesses int64 // Accepted deliveries (requests answered in time with ≥1 pool-accepted tx) RequestTimeouts int64 // Requests that timed out - LastLatencySample time.Time // Wall-clock time of the most recent request result (for staleness gate) + LatencyActivity float64 // EMA of accepted-delivery samples per block (eligibility gate for latency protection) } // peerStats is the internal mutable state per peer. @@ -80,7 +90,8 @@ type peerStats struct { requestLatencyEMA time.Duration requestSuccesses int64 requestTimeouts int64 - lastLatencySample time.Time + latencyActivity float64 + pendingSamples int // accepted-delivery samples since the last NotifyBlock } // Stats is the per-peer quality aggregator. @@ -126,15 +137,37 @@ func (s *Stats) NotifyBlock(inclusions, finalized map[string]int) { for peer, ps := range s.peers { ps.recentIncluded = (1-emaAlpha)*ps.recentIncluded + emaAlpha*float64(inclusions[peer]) ps.recentFinalized = (1-finalizedEMAAlpha)*ps.recentFinalized + finalizedEMAAlpha*float64(finalized[peer]) + + // Fold the accepted-delivery samples gathered since the previous + // head into the activity rate, then let it decay like the other + // per-block EMAs. + ps.latencyActivity = (1-latencyActivityAlpha)*ps.latencyActivity + latencyActivityAlpha*float64(ps.pendingSamples) + ps.pendingSamples = 0 + + // A peer silent long enough for its activity to fully decay + // forgets its latency history: a frozen fast EMA from a past + // active period must not be re-armable by rebuilding activity + // alone (see latencyResetThreshold). Gated on success history — + // only successes create a fast EMA worth forgetting; a + // timeout-only peer (activity permanently zero) keeps its + // penalty record. + if ps.latencyActivity < latencyResetThreshold && ps.requestSuccesses != 0 { + ps.latencyActivity = 0 + ps.requestLatencyEMA = 0 + ps.requestSuccesses = 0 + ps.requestTimeouts = 0 + } } } // NotifyRequestResult records a tx-request outcome for the given peer. // latency is the round-trip time (for timeouts, pass the timeout value). -// timeout indicates whether the request timed out rather than receiving a -// normal delivery. Both cases update the latency EMA; the timeout flag -// additionally increments the per-peer timeout counter. -// Creates a peer entry if one doesn't exist. +// timeout indicates whether the request timed out rather than receiving an +// accepted delivery (the fetcher only reports non-timeout results for +// deliveries with ≥1 pool-accepted tx). Both cases update the latency EMA; +// only accepted deliveries feed the activity rate that gates protection — +// a peer cannot become protection-eligible by timing out, and penalties +// remain ungated. Creates a peer entry if one doesn't exist. func (s *Stats) NotifyRequestResult(peer string, latency time.Duration, timeout bool) { s.mu.Lock() defer s.mu.Unlock() @@ -158,8 +191,8 @@ func (s *Stats) NotifyRequestResult(peer string, latency time.Duration, timeout ps.requestTimeouts++ } else { ps.requestSuccesses++ + ps.pendingSamples++ } - ps.lastLatencySample = time.Now() } // NotifyPeerDrop removes a peer's stats on disconnect. @@ -205,7 +238,7 @@ func (s *Stats) GetAllPeerStats() map[string]PeerStats { RequestLatencyEMA: ps.requestLatencyEMA, RequestSuccesses: ps.requestSuccesses, RequestTimeouts: ps.requestTimeouts, - LastLatencySample: ps.lastLatencySample, + LatencyActivity: ps.latencyActivity, } } return result diff --git a/eth/peerstats/peerstats_test.go b/eth/peerstats/peerstats_test.go index 4a46155a9d..a3fa8bdeb0 100644 --- a/eth/peerstats/peerstats_test.go +++ b/eth/peerstats/peerstats_test.go @@ -217,7 +217,8 @@ func TestPruneEmptyKeepClearsAll(t *testing.T) { // 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. +// dropper's MinLatencyActivity guard ensures this is harmless, and the +// dropper's periodic Prune reclaims the orphan. func TestStaleRequestLatencyAfterDrop(t *testing.T) { s := New() s.NotifyRequestResult("peerA", 200*time.Millisecond, false) @@ -232,7 +233,7 @@ func TestStaleRequestLatencyAfterDrop(t *testing.T) { if ps.RequestLatencyEMA != 50*time.Millisecond { t.Fatalf("expected fresh bootstrap at 50ms, got %v", ps.RequestLatencyEMA) } - // The dropper's MinLatencySamples guard (in eth/dropper.go) prevents + // The dropper's MinLatencyActivity guard (in eth/dropper.go) prevents // this 1-sample entry from earning latency-based protection. } @@ -258,34 +259,92 @@ func TestMultiplePeersIsolated(t *testing.T) { } } -// TestLatencyTimestampSet verifies that NotifyRequestResult stamps the -// peer's LastLatencySample with approximately time.Now(). -func TestLatencyTimestampSet(t *testing.T) { +// TestLatencyActivityAccumulatesAndDecays verifies that accepted-delivery +// samples fold into the activity EMA at the next NotifyBlock, that the +// pending counter resets after folding, and that subsequent sample-free +// blocks decay the activity. +func TestLatencyActivityAccumulatesAndDecays(t *testing.T) { s := New() - before := time.Now() - s.NotifyRequestResult("peerA", 100*time.Millisecond, false) - after := time.Now() + for i := 0; i < 10; i++ { + s.NotifyRequestResult("peerA", 100*time.Millisecond, false) + } + s.NotifyBlock(nil, nil) - got := s.GetAllPeerStats()["peerA"].LastLatencySample - if got.Before(before) || got.After(after) { - t.Fatalf("LastLatencySample = %v not in [%v, %v]", got, before, after) + folded := s.GetAllPeerStats()["peerA"].LatencyActivity + if folded <= 0 { + t.Fatalf("expected positive activity after folding samples, got %f", folded) + } + + // An empty block: nothing pending (counter was reset), pure decay. + s.NotifyBlock(nil, nil) + decayed := s.GetAllPeerStats()["peerA"].LatencyActivity + if decayed >= folded { + t.Fatalf("expected activity to decay on empty block, got %f >= %f", decayed, folded) + } + if decayed <= 0 { + t.Fatalf("expected gradual decay, not reset, got %f", decayed) } } -// TestLatencyTimestampUpdatesOnEachSample verifies that a later -// NotifyRequestResult call advances LastLatencySample. -func TestLatencyTimestampUpdatesOnEachSample(t *testing.T) { +// TestLatencyActivityGateReachable verifies that a peer sustaining one +// accepted delivery per block crosses MinLatencyActivity within a +// reasonable number of blocks (steady state for 1/block is 1.0). +func TestLatencyActivityGateReachable(t *testing.T) { s := New() - s.NotifyRequestResult("peerA", 100*time.Millisecond, false) - first := s.GetAllPeerStats()["peerA"].LastLatencySample + for i := 0; i < 20; i++ { + s.NotifyRequestResult("peerA", 100*time.Millisecond, false) + s.NotifyBlock(nil, nil) + } + if got := s.GetAllPeerStats()["peerA"].LatencyActivity; got < MinLatencyActivity { + t.Fatalf("sustained 1 sample/block should reach eligibility, got %f < %f", got, MinLatencyActivity) + } +} - // Small sleep so the second timestamp is detectably later. - time.Sleep(2 * time.Millisecond) - s.NotifyRequestResult("peerA", 200*time.Millisecond, false) - second := s.GetAllPeerStats()["peerA"].LastLatencySample +// TestTimeoutDoesNotFeedActivity verifies that timeouts update the EMA and +// counters but never contribute to the activity rate — a peer cannot become +// protection-eligible by timing out. +func TestTimeoutDoesNotFeedActivity(t *testing.T) { + s := New() + for i := 0; i < 50; i++ { + s.NotifyRequestResult("peerA", 5*time.Second, true) + s.NotifyBlock(nil, nil) + } + ps := s.GetAllPeerStats()["peerA"] + if ps.LatencyActivity != 0 { + t.Fatalf("timeouts must not feed activity, got %f", ps.LatencyActivity) + } + if ps.RequestTimeouts == 0 { + t.Fatal("expected timeout counter to advance") + } +} - if !second.After(first) { - t.Fatalf("expected second sample timestamp > first, got first=%v second=%v", first, second) +// TestLatencyStateForgottenAfterSilence verifies that once a silent peer's +// activity fully decays, its latency state (EMA and counters) is reset — +// a frozen fast EMA from a past active period cannot be re-armed later by +// rebuilding activity alone. +func TestLatencyStateForgottenAfterSilence(t *testing.T) { + s := New() + s.NotifyRequestResult("peerA", 50*time.Millisecond, false) + s.NotifyBlock(nil, nil) + if s.GetAllPeerStats()["peerA"].RequestLatencyEMA != 50*time.Millisecond { + t.Fatal("expected EMA seeded before silence") + } + + // Enough empty blocks for the activity to decay below the reset + // threshold (~200 blocks from a single sample at alpha=0.014). + for i := 0; i < 400; i++ { + s.NotifyBlock(nil, nil) + } + + ps := s.GetAllPeerStats()["peerA"] + if ps.RequestLatencyEMA != 0 || ps.RequestSuccesses != 0 || ps.RequestTimeouts != 0 || ps.LatencyActivity != 0 { + t.Fatalf("expected latency state forgotten after long silence, got %+v", ps) + } + + // A returning peer starts over: the next sample re-seeds the EMA. + s.NotifyRequestResult("peerA", 300*time.Millisecond, false) + if got := s.GetAllPeerStats()["peerA"].RequestLatencyEMA; got != 300*time.Millisecond { + t.Fatalf("expected fresh bootstrap after reset, got %v", got) } }