eth/peerstats: replace sample-count and staleness gates with activity EMA

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

@@ -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
+		// Low-latency peers rank higher. Eligibility requires a sustained
-		// score 0 so the existing `score > 0` filter excludes them — this
+		// rate of accepted-delivery samples (block-decayed EMA): scoring 0
-		// prevents a single lucky-fast reply from winning protection. Peers
+		// here lets the `score > 0` filter exclude peers whose EMA rests on
-		// whose EMA reaches the timeout also score low by this path because
+		// too little, too old, or front-loaded evidence — eligibility
-		// the reciprocal of a very large duration is tiny but positive; the
+		// expires on its own when the useful work stops. Peers whose EMA
-		// per-pool top-N will still push faster peers ahead of them.
+		// approaches the fetch timeout score tiny-but-positive via the
-		if s.RequestSuccesses+s.RequestTimeouts < peerstats.MinLatencySamples {
+		// reciprocal; per-pool top-N pushes faster peers ahead of them.
-			return 0
+		if s.LatencyActivity < peerstats.MinLatencyActivity {
-		}
-		// 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 {
 			return 0
 		}
 		if s.RequestLatencyEMA <= 0 {

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,
+		LatencyActivity:   peerstats.MinLatencyActivity,
-		LastLatencySample: time.Now(),
 	}
 	stats[dialed[1].ID().String()] = peerstats.PeerStats{
 		RequestLatencyEMA: 100 * time.Millisecond,
-		RequestSuccesses:  peerstats.MinLatencySamples,
+		LatencyActivity:   peerstats.MinLatencyActivity,
-		LastLatencySample: time.Now(),
 	}
 	stats[dialed[2].ID().String()] = peerstats.PeerStats{
 		RequestLatencyEMA: 2 * time.Second,
-		RequestSuccesses:  peerstats.MinLatencySamples,
+		LatencyActivity:   peerstats.MinLatencyActivity,
-		LastLatencySample: time.Now(),
 	}
 
 	protected := protectedPeersByPool(nil, dialed, stats)
@@ -273,22 +270,23 @@ func TestProtectedByPoolRequestLatencyBasic(t *testing.T) {
 	}
 }
 
-// TestProtectedByPoolRequestLatencyBootstrapGuard verifies that peers with
+// TestProtectedByPoolRequestLatencyBootstrapGuard verifies that peers whose
-// fewer than MinLatencySamples do not earn latency-based protection, even
+// accepted-delivery activity rate is below MinLatencyActivity do not earn
-// if their few samples indicate very low latency.
+// 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,
+		LatencyActivity:   peerstats.MinLatencyActivity,
-		LastLatencySample: time.Now(),
 	}
 
 	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,
+			LatencyActivity:   peerstats.MinLatencyActivity,
-			LastLatencySample: time.Now(),
 		}
 	}
 	// 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,
+			LatencyActivity:   peerstats.MinLatencyActivity,
-			LastLatencySample: time.Now(),
 		}
 	}
 
@@ -345,34 +341,33 @@ func TestProtectedByPoolRequestLatencyPerPool(t *testing.T) {
 	}
 }
 
-// TestProtectedByPoolRequestLatencyStale verifies that the freshness gate
+// TestProtectedByPoolRequestLatencyStale verifies that decayed activity
-// excludes peers whose latency EMA is valid (meeting the sample count and
+// excludes peers whose latency EMA is fast but whose accepted-delivery
-// fast value) but whose last sample is older than MaxLatencyStaleness.
+// rate has since fallen below MinLatencyActivity. A peer cannot serve a
-// A peer cannot serve a burst of fast replies, go silent on announcements,
+// burst of fast replies, go silent on announcements, and keep
-// and keep latency-based protection indefinitely.
+// 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,
+		LatencyActivity:   peerstats.MinLatencyActivity,
-		LastLatencySample: time.Now(),
 	}
-	// Stale, fast peer — was fast, but hasn't answered in too long.
+	// Formerly active, fast peer — same EMA, but its activity has decayed
-	// Same EMA and sample count as the fresh peer; only staleness differs.
+	// below the eligibility threshold.
 	stats[dialed[1].ID().String()] = peerstats.PeerStats{
 		RequestLatencyEMA: 50 * time.Millisecond,
-		RequestSuccesses:  peerstats.MinLatencySamples,
+		RequestSuccesses:  100,
-		LastLatencySample: time.Now().Add(-2 * peerstats.MaxLatencyStaleness),
+		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")
 	}
 }

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
+//     slow peers contribute to the EMA. Non-timeout results are only
-//     the per-peer timeout counter
+//     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
+	// EMA smoothing factor for the per-block latency-sample activity rate.
-	// before its RequestLatencyEMA is considered meaningful for protection.
+	// Half-life ≈ 50 chain heads (~10 minutes on 12s blocks): eligibility
-	// Prevents a single lucky-fast reply from displacing established peers.
+	// for latency protection must be continuously maintained at roughly
-	MinLatencySamples = 100
+	// this cadence, it cannot be front-loaded in a burst and then held.
-	// MaxLatencyStaleness is the oldest allowed age of a peer's last
+	latencyActivityAlpha = 0.014
-	// latency sample before their RequestLatencyEMA is disregarded for
+	// MinLatencyActivity is the minimum sustained rate of accepted-delivery
-	// protection. Prevents a peer from earning a fast score during a
+	// latency samples (per block, EMA-smoothed) a peer must maintain for its
-	// burst of activity and then holding protection indefinitely by
+	// RequestLatencyEMA to be considered for protection. 0.2 ≈ one accepted
-	// going silent on tx announcements (no further requests → no fresh
+	// fetch per five blocks (~1/minute). Replaces both an absolute sample
-	// samples → EMA frozen at its last value).
+	// count (front-loadable) and a last-sample staleness check (maintainable
-	MaxLatencyStaleness = 10 * time.Minute
+	// 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
+// timeout indicates whether the request timed out rather than receiving an
-// normal delivery. Both cases update the latency EMA; the timeout flag
+// accepted delivery (the fetcher only reports non-timeout results for
-// additionally increments the per-peer timeout counter.
+// deliveries with ≥1 pool-accepted tx). Both cases update the latency EMA;
-// Creates a peer entry if one doesn't exist.
+// 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

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
+// TestLatencyActivityAccumulatesAndDecays verifies that accepted-delivery
-// peer's LastLatencySample with approximately time.Now().
+// samples fold into the activity EMA at the next NotifyBlock, that the
-func TestLatencyTimestampSet(t *testing.T) {
+// pending counter resets after folding, and that subsequent sample-free
+// blocks decay the activity.
+func TestLatencyActivityAccumulatesAndDecays(t *testing.T) {
 	s := New()
-	before := time.Now()
+	for i := 0; i < 10; i++ {
 		s.NotifyRequestResult("peerA", 100*time.Millisecond, false)
-	after := time.Now()
+	}
+	s.NotifyBlock(nil, nil)
 
-	got := s.GetAllPeerStats()["peerA"].LastLatencySample
+	folded := s.GetAllPeerStats()["peerA"].LatencyActivity
-	if got.Before(before) || got.After(after) {
+	if folded <= 0 {
-		t.Fatalf("LastLatencySample = %v not in [%v, %v]", got, before, after)
+		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
+// TestLatencyActivityGateReachable verifies that a peer sustaining one
-// NotifyRequestResult call advances LastLatencySample.
+// accepted delivery per block crosses MinLatencyActivity within a
-func TestLatencyTimestampUpdatesOnEachSample(t *testing.T) {
+// reasonable number of blocks (steady state for 1/block is 1.0).
+func TestLatencyActivityGateReachable(t *testing.T) {
 	s := New()
+	for i := 0; i < 20; i++ {
 		s.NotifyRequestResult("peerA", 100*time.Millisecond, false)
-	first := s.GetAllPeerStats()["peerA"].LastLatencySample
+		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.
+// TestTimeoutDoesNotFeedActivity verifies that timeouts update the EMA and
-	time.Sleep(2 * time.Millisecond)
+// counters but never contribute to the activity rate — a peer cannot become
-	s.NotifyRequestResult("peerA", 200*time.Millisecond, false)
+// protection-eligible by timing out.
-	second := s.GetAllPeerStats()["peerA"].LastLatencySample
+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) {
+// TestLatencyStateForgottenAfterSilence verifies that once a silent peer's
-		t.Fatalf("expected second sample timestamp > first, got first=%v second=%v", first, second)
+// 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)
 	}
 }