Merge 91fed6db6b into dddbaa4bf3

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.txTracker.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)

eth/dropper.go

@@ -25,7 +25,7 @@ import (
 
 	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/common/mclock"
-	"github.com/ethereum/go-ethereum/eth/txtracker"
+	"github.com/ethereum/go-ethereum/eth/peerstats"
 	"github.com/ethereum/go-ethereum/log"
 	"github.com/ethereum/go-ethereum/metrics"
 	"github.com/ethereum/go-ethereum/p2p"
@@ -60,20 +60,36 @@ var (
 )
 
 // Callback type to get per-peer inclusion statistics.
-type getPeerStatsFunc func() map[string]txtracker.PeerStats
+type getPeerStatsFunc func() map[string]peerstats.PeerStats
 
 // protectionCategory defines a peer scoring function and the fraction of peers
 // to protect per inbound/dialed category. Multiple categories are unioned.
 type protectionCategory struct {
-	score func(txtracker.PeerStats) float64
+	score func(peerstats.PeerStats) float64
 	frac  float64 // fraction of max peers to protect (0.0–1.0)
 }
 
 // protectionCategories is the list of protection criteria. Each category
 // independently selects its top-N peers per pool; the union is protected.
 var protectionCategories = []protectionCategory{
-	{func(s txtracker.PeerStats) float64 { return s.RecentFinalized }, inclusionProtectionFrac}, // Recent finalized
-	{func(s txtracker.PeerStats) float64 { return s.RecentIncluded }, inclusionProtectionFrac},  // Recent included
+	{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 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 {
+			return 0
+		}
+		return 1.0 / float64(s.RequestLatencyEMA)
+	}, inclusionProtectionFrac},
 }
 
 // dropper monitors the state of the peer pool and introduces churn by
@@ -101,7 +117,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
@@ -131,12 +148,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()
 }
@@ -196,6 +214,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.
@@ -228,7 +261,7 @@ func (cm *dropper) protectedPeers(peers []*p2p.Peer) map[*p2p.Peer]bool {
 // Factored from protectedPeers so tests can exercise the per-pool
 // selection logic without needing to construct direction-flagged
 // *p2p.Peer instances (which require unexported p2p types).
-func protectedPeersByPool(inbound, dialed []*p2p.Peer, stats map[string]txtracker.PeerStats) map[*p2p.Peer]bool {
+func protectedPeersByPool(inbound, dialed []*p2p.Peer, stats map[string]peerstats.PeerStats) map[*p2p.Peer]bool {
 	result := make(map[*p2p.Peer]bool)
 	// protectPool selects the top-frac peers from pool by score and adds them to result.
 	protectPool := func(pool []*p2p.Peer, cat protectionCategory) {
@@ -274,6 +307,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()

eth/dropper_test.go

@@ -19,8 +19,9 @@ package eth
 import (
 	"fmt"
 	"testing"
+	"time"
 
-	"github.com/ethereum/go-ethereum/eth/txtracker"
+	"github.com/ethereum/go-ethereum/eth/peerstats"
 	"github.com/ethereum/go-ethereum/p2p"
 	"github.com/ethereum/go-ethereum/p2p/enode"
 )
@@ -36,7 +37,7 @@ func makePeers(n int) []*p2p.Peer {
 
 func TestProtectedPeersNoStats(t *testing.T) {
 	cm := &dropper{maxDialPeers: 20, maxInboundPeers: 30}
-	cm.peerStatsFunc = func() map[string]txtracker.PeerStats { return nil }
+	cm.peerStatsFunc = func() map[string]peerstats.PeerStats { return nil }
 
 	peers := makePeers(10)
 	protected := cm.protectedPeers(peers)
@@ -47,8 +48,8 @@ func TestProtectedPeersNoStats(t *testing.T) {
 
 func TestProtectedPeersEmptyStats(t *testing.T) {
 	cm := &dropper{maxDialPeers: 20, maxInboundPeers: 30}
-	cm.peerStatsFunc = func() map[string]txtracker.PeerStats {
-		return map[string]txtracker.PeerStats{}
+	cm.peerStatsFunc = func() map[string]peerstats.PeerStats {
+		return map[string]peerstats.PeerStats{}
 	}
 
 	peers := makePeers(10)
@@ -63,11 +64,11 @@ func TestProtectedPeersTopPeer(t *testing.T) {
 	cm := &dropper{maxDialPeers: 20, maxInboundPeers: 30}
 
 	peers := makePeers(20)
-	stats := make(map[string]txtracker.PeerStats)
-	stats[peers[0].ID().String()] = txtracker.PeerStats{RecentFinalized: 100}
-	stats[peers[1].ID().String()] = txtracker.PeerStats{RecentIncluded: 5.0}
+	stats := make(map[string]peerstats.PeerStats)
+	stats[peers[0].ID().String()] = peerstats.PeerStats{RecentFinalized: 100}
+	stats[peers[1].ID().String()] = peerstats.PeerStats{RecentIncluded: 5.0}
 
-	cm.peerStatsFunc = func() map[string]txtracker.PeerStats { return stats }
+	cm.peerStatsFunc = func() map[string]peerstats.PeerStats { return stats }
 
 	protected := cm.protectedPeers(peers)
 	if len(protected) != 2 {
@@ -85,11 +86,11 @@ func TestProtectedPeersZeroScore(t *testing.T) {
 	cm := &dropper{maxDialPeers: 20, maxInboundPeers: 30}
 
 	peers := makePeers(10)
-	stats := make(map[string]txtracker.PeerStats)
+	stats := make(map[string]peerstats.PeerStats)
 	for _, p := range peers {
-		stats[p.ID().String()] = txtracker.PeerStats{}
+		stats[p.ID().String()] = peerstats.PeerStats{}
 	}
-	cm.peerStatsFunc = func() map[string]txtracker.PeerStats { return stats }
+	cm.peerStatsFunc = func() map[string]peerstats.PeerStats { return stats }
 
 	protected := cm.protectedPeers(peers)
 	if len(protected) != 0 {
@@ -102,10 +103,10 @@ func TestProtectedPeersOverlap(t *testing.T) {
 	cm := &dropper{maxDialPeers: 20, maxInboundPeers: 30}
 
 	peers := makePeers(20)
-	stats := make(map[string]txtracker.PeerStats)
-	stats[peers[0].ID().String()] = txtracker.PeerStats{RecentFinalized: 100, RecentIncluded: 5.0}
+	stats := make(map[string]peerstats.PeerStats)
+	stats[peers[0].ID().String()] = peerstats.PeerStats{RecentFinalized: 100, RecentIncluded: 5.0}
 
-	cm.peerStatsFunc = func() map[string]txtracker.PeerStats { return stats }
+	cm.peerStatsFunc = func() map[string]peerstats.PeerStats { return stats }
 
 	protected := cm.protectedPeers(peers)
 	if len(protected) != 1 {
@@ -138,12 +139,12 @@ func TestProtectedByPoolPerPoolTopN(t *testing.T) {
 		dialed[i] = p2p.NewPeer(id, fmt.Sprintf("dialed%d", i), nil)
 	}
 	// Strictly increasing scores: highest wins in each pool.
-	stats := make(map[string]txtracker.PeerStats)
+	stats := make(map[string]peerstats.PeerStats)
 	for i, p := range inbound {
-		stats[p.ID().String()] = txtracker.PeerStats{RecentFinalized: float64(1 + i)}
+		stats[p.ID().String()] = peerstats.PeerStats{RecentFinalized: float64(1 + i)}
 	}
 	for i, p := range dialed {
-		stats[p.ID().String()] = txtracker.PeerStats{RecentFinalized: float64(1 + i)}
+		stats[p.ID().String()] = peerstats.PeerStats{RecentFinalized: float64(1 + i)}
 	}
 
 	protected := protectedPeersByPool(inbound, dialed, stats)
@@ -173,10 +174,10 @@ func TestProtectedByPoolCrossCategoryOverlap(t *testing.T) {
 	//   RecentFinalized winners: P2 (tie-broken-ok), P0
 	//   RecentIncluded winners: P2, P1
 	// Union: {P0, P1, P2}.
-	stats := make(map[string]txtracker.PeerStats)
-	stats[dialed[0].ID().String()] = txtracker.PeerStats{RecentFinalized: 100, RecentIncluded: 0}
-	stats[dialed[1].ID().String()] = txtracker.PeerStats{RecentFinalized: 0, RecentIncluded: 5.0}
-	stats[dialed[2].ID().String()] = txtracker.PeerStats{RecentFinalized: 200, RecentIncluded: 10.0}
+	stats := make(map[string]peerstats.PeerStats)
+	stats[dialed[0].ID().String()] = peerstats.PeerStats{RecentFinalized: 100, RecentIncluded: 0}
+	stats[dialed[1].ID().String()] = peerstats.PeerStats{RecentFinalized: 0, RecentIncluded: 5.0}
+	stats[dialed[2].ID().String()] = peerstats.PeerStats{RecentFinalized: 200, RecentIncluded: 10.0}
 
 	protected := protectedPeersByPool(nil, dialed, stats)
 
@@ -203,13 +204,13 @@ func TestProtectedByPoolPerPoolIndependence(t *testing.T) {
 		id := enode.ID{byte(100 + i)}
 		dialed[i] = p2p.NewPeer(id, fmt.Sprintf("dialed%d", i), nil)
 	}
-	stats := make(map[string]txtracker.PeerStats)
+	stats := make(map[string]peerstats.PeerStats)
 	// Every inbound peer outscores every dialed peer.
 	for i, p := range inbound {
-		stats[p.ID().String()] = txtracker.PeerStats{RecentFinalized: float64(1000 + i)}
+		stats[p.ID().String()] = peerstats.PeerStats{RecentFinalized: float64(1000 + i)}
 	}
 	for i, p := range dialed {
-		stats[p.ID().String()] = txtracker.PeerStats{RecentFinalized: float64(1 + i)}
+		stats[p.ID().String()] = peerstats.PeerStats{RecentFinalized: float64(1 + i)}
 	}
 
 	protected := protectedPeersByPool(inbound, dialed, stats)
@@ -232,3 +233,141 @@ func TestProtectedByPoolPerPoolIndependence(t *testing.T) {
 		t.Fatalf("expected 4 protected peers (top-2 of each pool), got %d", len(protected))
 	}
 }
+
+// TestProtectedByPoolRequestLatencyBasic verifies the latency protection
+// category: with no competing inclusion stats, the lowest-latency peers
+// (among those with enough samples) win top-N protection.
+func TestProtectedByPoolRequestLatencyBasic(t *testing.T) {
+	dialed := makePeers(20) // frac=0.1 → n=2 per category
+	stats := make(map[string]peerstats.PeerStats)
+	// Three peers have enough samples; the two fastest should win.
+	stats[dialed[0].ID().String()] = peerstats.PeerStats{
+		RequestLatencyEMA: 50 * time.Millisecond,
+		LatencyActivity:   peerstats.MinLatencyActivity,
+	}
+	stats[dialed[1].ID().String()] = peerstats.PeerStats{
+		RequestLatencyEMA: 100 * time.Millisecond,
+		LatencyActivity:   peerstats.MinLatencyActivity,
+	}
+	stats[dialed[2].ID().String()] = peerstats.PeerStats{
+		RequestLatencyEMA: 2 * time.Second,
+		LatencyActivity:   peerstats.MinLatencyActivity,
+	}
+
+	protected := protectedPeersByPool(nil, dialed, stats)
+
+	if !protected[dialed[0]] {
+		t.Error("fastest peer should be protected")
+	}
+	if !protected[dialed[1]] {
+		t.Error("second-fastest peer should be protected")
+	}
+	if protected[dialed[2]] {
+		t.Error("slowest peer should not be in top-2")
+	}
+	if len(protected) != 2 {
+		t.Fatalf("expected top-2 latency protection, got %d", len(protected))
+	}
+}
+
+// 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 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,
+		LatencyActivity:   peerstats.MinLatencyActivity,
+	}
+
+	protected := protectedPeersByPool(nil, dialed, stats)
+
+	if protected[dialed[0]] {
+		t.Error("under-sampled peer should not be protected (bootstrap guard)")
+	}
+	if !protected[dialed[1]] {
+		t.Error("warmed-up peer should be protected")
+	}
+}
+
+// TestProtectedByPoolRequestLatencyPerPool verifies that the latency
+// category selects top-N per pool independently, consistent with the
+// other categories. An inbound peer with lower latency does not prevent
+// a dialed peer from being protected as top of the dialed pool.
+func TestProtectedByPoolRequestLatencyPerPool(t *testing.T) {
+	inbound := makePeers(20)
+	dialed := make([]*p2p.Peer, 20)
+	for i := range dialed {
+		id := enode.ID{byte(100 + i)}
+		dialed[i] = p2p.NewPeer(id, fmt.Sprintf("dialed%d", i), nil)
+	}
+	stats := make(map[string]peerstats.PeerStats)
+	// All inbound peers are very fast (50ms).
+	for _, p := range inbound {
+		stats[p.ID().String()] = peerstats.PeerStats{
+			RequestLatencyEMA: 50 * time.Millisecond,
+			LatencyActivity:   peerstats.MinLatencyActivity,
+		}
+	}
+	// Dialed peers are slower (1s) — globally they would all lose, but
+	// per-pool top-N should still protect two of them.
+	for _, p := range dialed {
+		stats[p.ID().String()] = peerstats.PeerStats{
+			RequestLatencyEMA: 1 * time.Second,
+			LatencyActivity:   peerstats.MinLatencyActivity,
+		}
+	}
+
+	protected := protectedPeersByPool(inbound, dialed, stats)
+
+	// 2 from inbound + 2 from dialed = 4.
+	var dialedProtected int
+	for _, p := range dialed {
+		if protected[p] {
+			dialedProtected++
+		}
+	}
+	if dialedProtected != 2 {
+		t.Fatalf("expected 2 dialed peers protected by per-pool top-N, got %d", dialedProtected)
+	}
+}
+
+// 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)
+	// Active, fast peer — should be protected.
+	stats[dialed[0].ID().String()] = peerstats.PeerStats{
+		RequestLatencyEMA: 50 * time.Millisecond,
+		LatencyActivity:   peerstats.MinLatencyActivity,
+	}
+	// 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:  100,
+		LatencyActivity:   peerstats.MinLatencyActivity * 0.9,
+	}
+
+	protected := protectedPeersByPool(nil, dialed, stats)
+
+	if !protected[dialed[0]] {
+		t.Error("active fast peer must be protected")
+	}
+	if protected[dialed[1]] {
+		t.Error("decayed-activity peer must NOT keep latency protection despite fast EMA")
+	}
+}

eth/fetcher/tx_fetcher.go

@@ -127,6 +127,7 @@ type txDelivery struct {
 	hashes    []common.Hash // Batch of transaction hashes having been delivered
 	metas     []txMetadata  // Batch of metadata associated with the delivered hashes
 	direct    bool          // Whether this is a direct reply or a broadcast
+	accepted  bool          // Whether any delivered tx was newly accepted by the pool
 	violation error         // Whether we encountered a protocol violation
 }
 
@@ -183,11 +184,12 @@ type TxFetcher struct {
 	alternates map[common.Hash]map[string]struct{} // In-flight transaction alternate origins if retrieval fails
 
 	// Callbacks
-	validateMeta func(common.Hash, byte) error           // Validate a tx metadata based on the local txpool
-	addTxs       func([]*types.Transaction) []error      // Insert a batch of transactions into local txpool
-	fetchTxs     func(string, []common.Hash) error       // Retrieves a set of txs from a remote peer
-	dropPeer     func(string)                            // Drops a peer in case of announcement violation
-	onAccepted   func(peer string, hashes []common.Hash) // Optional: notified with accepted tx hashes per peer
+	validateMeta    func(common.Hash, byte) error                          // Validate a tx metadata based on the local txpool
+	addTxs          func([]*types.Transaction) []error                     // Insert a batch of transactions into local txpool
+	fetchTxs        func(string, []common.Hash) error                      // Retrieves a set of txs from a remote peer
+	dropPeer        func(string)                                           // Drops a peer in case of announcement violation
+	onAccepted      func(peer string, hashes []common.Hash)                // Optional: notified with accepted tx hashes per peer
+	onRequestResult func(peer string, latency time.Duration, timeout bool) // Optional: notified per timed-out request and per delivery with pool-accepted txs
 
 	buffer *blobpool.BlobBuffer
 
@@ -201,41 +203,42 @@ type TxFetcher struct {
 // based on hash announcements.
 // Chain can be nil to disable on-chain checks.
 func NewTxFetcher(chain *core.BlockChain, validateMeta func(common.Hash, byte) error, addTxs func([]*types.Transaction) []error, fetchTxs func(string, []common.Hash) error,
-	dropPeer func(string), onAccepted func(string, []common.Hash), buffer *blobpool.BlobBuffer) *TxFetcher {
-	return NewTxFetcherForTests(chain, validateMeta, addTxs, fetchTxs, dropPeer, onAccepted, buffer, mclock.System{}, time.Now, nil)
+	dropPeer func(string), onAccepted func(string, []common.Hash), onRequestResult func(string, time.Duration, bool), buffer *blobpool.BlobBuffer) *TxFetcher {
+	return NewTxFetcherForTests(chain, validateMeta, addTxs, fetchTxs, dropPeer, onAccepted, onRequestResult, buffer, mclock.System{}, time.Now, nil)
 }
 
 // NewTxFetcherForTests is a testing method to mock out the realtime clock with
 // a simulated version and the internal randomness with a deterministic one.
 // Chain can be nil to disable on-chain checks.
 func NewTxFetcherForTests(
-	chain *core.BlockChain, validateMeta func(common.Hash, byte) error, addTxs func([]*types.Transaction) []error, fetchTxs func(string, []common.Hash) error, dropPeer func(string), onAccepted func(string, []common.Hash),
+	chain *core.BlockChain, validateMeta func(common.Hash, byte) error, addTxs func([]*types.Transaction) []error, fetchTxs func(string, []common.Hash) error, dropPeer func(string), onAccepted func(string, []common.Hash), onRequestResult func(string, time.Duration, bool),
 	buffer *blobpool.BlobBuffer, clock mclock.Clock, realTime func() time.Time, rand *mrand.Rand) *TxFetcher {
 	return &TxFetcher{
-		notify:         make(chan *txAnnounce),
-		cleanup:        make(chan *txDelivery),
-		drop:           make(chan *txDrop),
-		quit:           make(chan struct{}),
-		waitlist:       make(map[common.Hash]map[string]struct{}),
-		waittime:       make(map[common.Hash]mclock.AbsTime),
-		waitslots:      make(map[string]map[common.Hash]*txMetadataWithSeq),
-		announces:      make(map[string]map[common.Hash]*txMetadataWithSeq),
-		announced:      make(map[common.Hash]map[string]struct{}),
-		fetching:       make(map[common.Hash]string),
-		requests:       make(map[string]*txRequest),
-		alternates:     make(map[common.Hash]map[string]struct{}),
-		underpriced:    lru.NewCache[common.Hash, time.Time](maxTxUnderpricedSetSize),
-		txOnChainCache: lru.NewCache[common.Hash, struct{}](txOnChainCacheLimit),
-		chain:          chain,
-		validateMeta:   validateMeta,
-		addTxs:         addTxs,
-		fetchTxs:       fetchTxs,
-		dropPeer:       dropPeer,
-		buffer:         buffer,
-		onAccepted:     onAccepted,
-		clock:          clock,
-		realTime:       realTime,
-		rand:           rand,
+		notify:          make(chan *txAnnounce),
+		cleanup:         make(chan *txDelivery),
+		drop:            make(chan *txDrop),
+		quit:            make(chan struct{}),
+		waitlist:        make(map[common.Hash]map[string]struct{}),
+		waittime:        make(map[common.Hash]mclock.AbsTime),
+		waitslots:       make(map[string]map[common.Hash]*txMetadataWithSeq),
+		announces:       make(map[string]map[common.Hash]*txMetadataWithSeq),
+		announced:       make(map[common.Hash]map[string]struct{}),
+		fetching:        make(map[common.Hash]string),
+		requests:        make(map[string]*txRequest),
+		alternates:      make(map[common.Hash]map[string]struct{}),
+		underpriced:     lru.NewCache[common.Hash, time.Time](maxTxUnderpricedSetSize),
+		txOnChainCache:  lru.NewCache[common.Hash, struct{}](txOnChainCacheLimit),
+		chain:           chain,
+		validateMeta:    validateMeta,
+		addTxs:          addTxs,
+		fetchTxs:        fetchTxs,
+		dropPeer:        dropPeer,
+		buffer:          buffer,
+		onAccepted:      onAccepted,
+		onRequestResult: onRequestResult,
+		clock:           clock,
+		realTime:        realTime,
+		rand:            rand,
 	}
 }
 
@@ -355,8 +358,9 @@ func (f *TxFetcher) Enqueue(peer string, version uint, txs []*types.Transaction,
 	// Push all the transactions into the pool, tracking underpriced ones to avoid
 	// re-requesting them and dropping the peer in case of malicious transfers.
 	var (
-		added = make([]common.Hash, 0, len(txs))
-		metas = make([]txMetadata, 0, len(txs))
+		added       = make([]common.Hash, 0, len(txs))
+		metas       = make([]txMetadata, 0, len(txs))
+		anyAccepted bool
 	)
 	// proceed in batches
 	for i := 0; i < len(txs); i += addTxsBatchSize {
@@ -414,8 +418,11 @@ func (f *TxFetcher) Enqueue(peer string, version uint, txs []*types.Transaction,
 		otherreject := f.handleAddErrors(hashes, errs, metrics)
 
 		// Notify the tracker which txs from this peer were accepted.
-		if f.onAccepted != nil && len(accepted) > 0 {
-			f.onAccepted(peer, accepted)
+		if len(accepted) > 0 {
+			anyAccepted = true
+			if f.onAccepted != nil {
+				f.onAccepted(peer, accepted)
+			}
 		}
 		// If 'other reject' is >25% of the deliveries in any batch, sleep a bit
 		// to throttle the misbehaving peer.
@@ -429,7 +436,7 @@ func (f *TxFetcher) Enqueue(peer string, version uint, txs []*types.Transaction,
 		}
 	}
 	select {
-	case f.cleanup <- &txDelivery{origin: peer, hashes: added, metas: metas, direct: direct, violation: violation}:
+	case f.cleanup <- &txDelivery{origin: peer, hashes: added, metas: metas, direct: direct, accepted: anyAccepted, violation: violation}:
 		return nil
 	case <-f.quit:
 		return errTerminated
@@ -735,6 +742,14 @@ func (f *TxFetcher) loop() {
 					// Keep track of the request as dangling, but never expire
 					f.requests[peer].hashes = nil
 					txFetcherSlowPeers.Inc(1)
+					// Record the request as a timeout-latency sample. The slow
+					// EMA in the consumer counts timeouts as the timeout value
+					// itself, so a peer that times out repeatedly drags its
+					// score down without us having to wait for an eventual
+					// (possibly never-arriving) reply.
+					if f.onRequestResult != nil {
+						f.onRequestResult(peer, txFetchTimeout, true)
+					}
 				}
 			}
 			// Schedule a new transaction retrieval
@@ -831,6 +846,14 @@ func (f *TxFetcher) loop() {
 				if req.hashes == nil {
 					txFetcherSlowPeers.Dec(1)
 					txFetcherSlowWait.Update(time.Duration(f.clock.Now() - req.time).Nanoseconds())
+					// Already counted as a timeout sample at the timeout site;
+					// don't double-record on eventual delivery.
+				} else if f.onRequestResult != nil && delivery.accepted {
+					// In-time delivery that yielded at least one pool-accepted
+					// tx. Record the actual round-trip. Deliveries without any
+					// accepted tx (duplicates, rejects) record nothing — latency
+					// samples must not be farmable from worthless responses.
+					f.onRequestResult(delivery.origin, time.Duration(f.clock.Now()-req.time), false)
 				}
 				delete(f.requests, delivery.origin)
 

eth/fetcher/tx_fetcher_test.go

@@ -22,6 +22,7 @@ import (
 	"math/big"
 	"math/rand"
 	"slices"
+	"sync"
 	"testing"
 	"time"
 
@@ -112,6 +113,7 @@ func newTestTxFetcher() *TxFetcher {
 		func(string, []common.Hash) error { return nil },
 		nil,
 		nil,
+		nil,
 		newTestBlobBuffer(),
 	)
 }
@@ -2219,6 +2221,7 @@ func TestTransactionForgotten(t *testing.T) {
 		func(string, []common.Hash) error { return nil },
 		func(string) {},
 		nil,
+		nil,
 		newTestBlobBuffer(),
 		mockClock,
 		mockTime,
@@ -2300,3 +2303,141 @@ func TestTransactionForgotten(t *testing.T) {
 		t.Errorf("wrong final underpriced cache size: got %d, want 1", size)
 	}
 }
+
+// resultRecorder is a thread-safe recorder for onRequestResult callbacks.
+type resultRecorder struct {
+	mu      sync.Mutex
+	samples []resultSample
+}
+
+type resultSample struct {
+	peer    string
+	latency time.Duration
+	timeout bool
+}
+
+func (r *resultRecorder) record(peer string, latency time.Duration, timeout bool) {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	r.samples = append(r.samples, resultSample{peer, latency, timeout})
+}
+
+func (r *resultRecorder) snapshot() []resultSample {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	out := make([]resultSample, len(r.samples))
+	copy(out, r.samples)
+	return out
+}
+
+// TestTransactionFetcherRequestResultOnDelivery asserts that an in-time
+// direct delivery fires the onRequestResult callback with timeout=false.
+func TestTransactionFetcherRequestResultOnDelivery(t *testing.T) {
+	rec := &resultRecorder{}
+	testTransactionFetcherParallel(t, txFetcherTest{
+		init: func() *TxFetcher {
+			f := newTestTxFetcher()
+			f.onRequestResult = rec.record
+			return f
+		},
+		steps: []interface{}{
+			doTxNotify{peer: "A", hashes: []common.Hash{testTxsHashes[0]}, types: []byte{testTxs[0].Type()}, sizes: []uint32{uint32(testTxs[0].Size())}},
+			doWait{time: txArriveTimeout, step: true},
+			doWait{time: 200 * time.Millisecond, step: false},
+			doTxEnqueue{peer: "A", txs: []*types.Transaction{testTxs[0]}, direct: true},
+			doFunc(func() {
+				samples := rec.snapshot()
+				if len(samples) != 1 {
+					t.Fatalf("expected 1 sample, got %d (%v)", len(samples), samples)
+				}
+				if samples[0].peer != "A" {
+					t.Errorf("peer mismatch: got %q, want A", samples[0].peer)
+				}
+				if samples[0].latency != 200*time.Millisecond {
+					t.Errorf("latency mismatch: got %v, want 200ms", samples[0].latency)
+				}
+				if samples[0].timeout {
+					t.Error("expected timeout=false for delivery")
+				}
+			}),
+		},
+	})
+}
+
+// TestTransactionFetcherRequestResultOnTimeout asserts that a timed-out
+// request fires onRequestResult with timeout=true and the timeout value,
+// and a subsequent (late) delivery does not fire a duplicate sample.
+func TestTransactionFetcherRequestResultOnTimeout(t *testing.T) {
+	rec := &resultRecorder{}
+	testTransactionFetcherParallel(t, txFetcherTest{
+		init: func() *TxFetcher {
+			f := newTestTxFetcher()
+			f.onRequestResult = rec.record
+			return f
+		},
+		steps: []interface{}{
+			doTxNotify{peer: "A", hashes: []common.Hash{testTxsHashes[0]}, types: []byte{testTxs[0].Type()}, sizes: []uint32{uint32(testTxs[0].Size())}},
+			doWait{time: txArriveTimeout, step: true},
+			doWait{time: txFetchTimeout, step: true},
+			doFunc(func() {
+				samples := rec.snapshot()
+				if len(samples) != 1 {
+					t.Fatalf("expected 1 timeout sample, got %d (%v)", len(samples), samples)
+				}
+				if samples[0].peer != "A" {
+					t.Errorf("peer mismatch: got %q, want A", samples[0].peer)
+				}
+				if samples[0].latency != txFetchTimeout {
+					t.Errorf("latency mismatch: got %v, want %v", samples[0].latency, txFetchTimeout)
+				}
+				if !samples[0].timeout {
+					t.Error("expected timeout=true for timed-out request")
+				}
+			}),
+			doTxEnqueue{peer: "A", txs: []*types.Transaction{testTxs[0]}, direct: true},
+			doFunc(func() {
+				if len(rec.snapshot()) != 1 {
+					t.Fatalf("late delivery double-counted: got %d samples, want 1", len(rec.snapshot()))
+				}
+			}),
+		},
+	})
+}
+
+// TestTransactionFetcherRequestResultRequiresAcceptance asserts that an
+// in-time delivery containing no pool-accepted transactions (e.g. all
+// duplicates) does NOT record a latency sample — a peer cannot farm
+// latency protection by answering fetches with worthless content.
+func TestTransactionFetcherRequestResultRequiresAcceptance(t *testing.T) {
+	rec := &resultRecorder{}
+	testTransactionFetcherParallel(t, txFetcherTest{
+		init: func() *TxFetcher {
+			f := NewTxFetcher(
+				nil,
+				func(common.Hash, byte) error { return nil },
+				func(txs []*types.Transaction) []error {
+					errs := make([]error, len(txs))
+					for i := range errs {
+						errs[i] = txpool.ErrAlreadyKnown
+					}
+					return errs
+				},
+				func(string, []common.Hash) error { return nil },
+				nil, nil, rec.record,
+				newTestBlobBuffer(),
+			)
+			return f
+		},
+		steps: []interface{}{
+			doTxNotify{peer: "A", hashes: []common.Hash{testTxsHashes[0]}, types: []byte{testTxs[0].Type()}, sizes: []uint32{uint32(testTxs[0].Size())}},
+			doWait{time: txArriveTimeout, step: true},
+			doWait{time: 200 * time.Millisecond, step: false},
+			doTxEnqueue{peer: "A", txs: []*types.Transaction{testTxs[0]}, direct: true},
+			doFunc(func() {
+				if samples := rec.snapshot(); len(samples) != 0 {
+					t.Fatalf("expected no sample for unaccepted delivery, got %d (%v)", len(samples), samples)
+				}
+			}),
+		},
+	})
+}

eth/handler.go

@@ -38,6 +38,7 @@ import (
 	"github.com/ethereum/go-ethereum/eth/downloader"
 	"github.com/ethereum/go-ethereum/eth/ethconfig"
 	"github.com/ethereum/go-ethereum/eth/fetcher"
+	"github.com/ethereum/go-ethereum/eth/peerstats"
 	"github.com/ethereum/go-ethereum/eth/protocols/eth"
 	"github.com/ethereum/go-ethereum/eth/protocols/snap"
 	"github.com/ethereum/go-ethereum/eth/txtracker"
@@ -142,6 +143,7 @@ type handler struct {
 	txFetcher      *fetcher.TxFetcher
 	blobFetcher    *fetcher.BlobFetcher
 	txTracker      *txtracker.Tracker
+	peerStats      *peerstats.Stats
 	peers          *peerSet
 	txBroadcastKey [16]byte
 
@@ -211,7 +213,8 @@ func newHandler(config *handlerConfig) (*handler, error) {
 		return nil
 	}
 	h.txTracker = txtracker.New()
-	h.txFetcher = fetcher.NewTxFetcher(h.chain, validateMeta, addTxs, fetchTx, h.removePeer, h.txTracker.NotifyAccepted, blobBuffer)
+	h.peerStats = peerstats.New()
+	h.txFetcher = fetcher.NewTxFetcher(h.chain, validateMeta, addTxs, fetchTx, h.removePeer, h.txTracker.NotifyAccepted, h.peerStats.NotifyRequestResult, blobBuffer)
 
 	// Construct the blob fetcher for cell-based blob data availability
 	blobCallbacks := fetcher.BlobFetcherFunctions{
@@ -456,7 +459,7 @@ func (h *handler) unregisterPeer(id string) {
 	h.downloader.UnregisterPeer(id)
 	h.txFetcher.Drop(id)
 	h.blobFetcher.Drop(id)
-	h.txTracker.NotifyPeerDrop(id)
+	h.peerStats.NotifyPeerDrop(id)
 
 	if err := h.peers.unregisterPeer(id); err != nil {
 		logger.Error("Ethereum peer removal failed", "err", err)
@@ -481,8 +484,10 @@ func (h *handler) Start(maxPeers int) {
 	h.txFetcher.Start()
 	h.blobFetcher.Start()
 
-	// Start the transaction tracker (records tx deliveries, credits peer inclusions).
-	h.txTracker.Start(h.chain)
+	// Start the transaction tracker; it emits per-block inclusion and
+	// finalization signals to peerStats, which the dropper queries for
+	// protection decisions.
+	h.txTracker.Start(h.chain, h.peerStats)
 
 	// start peer handler tracker
 	h.wg.Add(1)

eth/peerstats/peerstats.go

@@ -0,0 +1,245 @@
+// Copyright 2026 The go-ethereum Authors
+// This file is part of the go-ethereum library.
+//
+// The go-ethereum library is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Lesser General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// The go-ethereum library is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU Lesser General Public License for more details.
+//
+// You should have received a copy of the GNU Lesser General Public License
+// along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
+
+// Package peerstats maintains per-peer quality metrics used by the peer
+// dropper to protect high-value peers from random disconnection.
+//
+// The package is a passive accumulator: it exposes entry points for its
+// signal producers (txtracker for inclusion/finalization, the tx fetcher
+// for latency, the handler for peer-drop cleanup) and a read-only
+// snapshot for its consumer (the dropper). It has no goroutine of its
+// own — all mutation is serialized by a single mutex.
+//
+// Signal sources:
+//   - NotifyBlock(inclusions, finalized) — per-block deltas from txtracker
+//     (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. 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
+
+import (
+	"sync"
+	"time"
+)
+
+const (
+	// EMA smoothing factor for per-block inclusion rate.
+	emaAlpha = 0.05
+	// EMA smoothing factor for per-block finalization rate. Very slow on
+	// purpose: finalization is permanent, and the score should reflect
+	// sustained contribution over long windows, not recent bursts.
+	// Half-life ≈ 6930 chain heads (~23 hours on 12s blocks).
+	finalizedEMAAlpha = 0.0001
+	// EMA smoothing factor for per-request latency average. Slow on purpose:
+	// short bursts shouldn't shift the score, sustained behavior should.
+	// Half-life ≈ ln(0.5)/ln(0.99) ≈ 69 samples.
+	latencyEMAAlpha = 0.01
+	// 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.
+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         // Accepted deliveries (requests answered in time with ≥1 pool-accepted tx)
+	RequestTimeouts   int64         // Requests that timed out
+	LatencyActivity   float64       // EMA of accepted-delivery samples per block (eligibility gate for latency protection)
+}
+
+// peerStats is the internal mutable state per peer.
+type peerStats struct {
+	recentFinalized   float64
+	recentIncluded    float64
+	requestLatencyEMA time.Duration
+	requestSuccesses  int64
+	requestTimeouts   int64
+	latencyActivity   float64
+	pendingSamples    int // accepted-delivery samples since the last NotifyBlock
+}
+
+// Stats is the per-peer quality aggregator.
+type Stats struct {
+	mu    sync.Mutex
+	peers map[string]*peerStats
+}
+
+// New creates an empty Stats.
+func New() *Stats {
+	return &Stats{peers: make(map[string]*peerStats)}
+}
+
+// NotifyBlock ingests a per-block update. `inclusions` is the count of the head
+// block's transactions attributed to each peer; peers with a positive
+// count get a stats entry created if one doesn't exist (this is how
+// peerstats learns about newly-active peers). Peers not in the map but
+// already tracked have their EMA decay with a zero sample.
+//
+// `finalized` is per-peer credits accumulated since the last NotifyBlock;
+// credits are only applied to peers already tracked — we don't resurrect
+// dropped peers from historical finalization data.
+//
+// NotifyBlock must NOT be called while the caller holds any other lock that
+// could be acquired by peerstats callers in reverse order. Current callers
+// (txtracker.handleChainHead) release their lock before invoking NotifyBlock.
+func (s *Stats) NotifyBlock(inclusions, finalized map[string]int) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+
+	// Ensure a stats entry exists for any peer that just had an inclusion.
+	// This is the primary path by which peerstats learns about a peer's
+	// inclusion activity.
+	for peer, count := range inclusions {
+		if count > 0 && s.peers[peer] == nil {
+			s.peers[peer] = &peerStats{}
+		}
+	}
+	// Update inclusion and finalization EMAs for every tracked peer. A
+	// peer not present in the respective delta map gets a 0 contribution
+	// — pure decay. Finalization credits for peers no longer tracked are
+	// ignored (don't resurrect dropped peers from historical data).
+	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 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()
+
+	ps := s.peers[peer]
+	if ps == nil {
+		ps = &peerStats{}
+		s.peers[peer] = ps
+	}
+	if ps.requestSuccesses+ps.requestTimeouts == 0 {
+		// Bootstrap the EMA with the first sample so it doesn't drift up
+		// from zero over many samples before reaching realistic values.
+		ps.requestLatencyEMA = latency
+	} else {
+		ps.requestLatencyEMA = time.Duration(
+			float64(ps.requestLatencyEMA)*(1-latencyEMAAlpha) +
+				float64(latency)*latencyEMAAlpha,
+		)
+	}
+	if timeout {
+		ps.requestTimeouts++
+	} else {
+		ps.requestSuccesses++
+		ps.pendingSamples++
+	}
+}
+
+// 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 {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+
+	result := make(map[string]PeerStats, len(s.peers))
+	for id, ps := range s.peers {
+		result[id] = PeerStats{
+			RecentFinalized:   ps.recentFinalized,
+			RecentIncluded:    ps.recentIncluded,
+			RequestLatencyEMA: ps.requestLatencyEMA,
+			RequestSuccesses:  ps.requestSuccesses,
+			RequestTimeouts:   ps.requestTimeouts,
+			LatencyActivity:   ps.latencyActivity,
+		}
+	}
+	return result
+}

eth/peerstats/peerstats_test.go

@@ -0,0 +1,384 @@
+// Copyright 2026 The go-ethereum Authors
+// This file is part of the go-ethereum library.
+//
+// The go-ethereum library is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Lesser General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// The go-ethereum library is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU Lesser General Public License for more details.
+//
+// You should have received a copy of the GNU Lesser General Public License
+// along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
+
+package peerstats
+
+import (
+	"testing"
+	"time"
+)
+
+// TestNotifyBlockBootstrapsFromInclusions verifies that a peer with a positive
+// inclusion count in the first NotifyBlock gets a stats entry created.
+func TestNotifyBlockBootstrapsFromInclusions(t *testing.T) {
+	s := New()
+	s.NotifyBlock(map[string]int{"peerA": 3}, nil)
+
+	stats := s.GetAllPeerStats()
+	if len(stats) != 1 {
+		t.Fatalf("expected 1 peer entry, got %d", len(stats))
+	}
+	ps, ok := stats["peerA"]
+	if !ok {
+		t.Fatal("expected peerA entry")
+	}
+	// EMA after first block: (1-0.05)*0 + 0.05*3 = 0.15
+	if ps.RecentIncluded <= 0 {
+		t.Fatalf("expected RecentIncluded > 0 after inclusion, got %f", ps.RecentIncluded)
+	}
+}
+
+// TestNotifyBlockDecaysKnownPeers verifies that peers already tracked get their
+// RecentIncluded EMA decayed when they have no inclusions in a block.
+func TestNotifyBlockDecaysKnownPeers(t *testing.T) {
+	s := New()
+	// Seed peerA with an inclusion.
+	s.NotifyBlock(map[string]int{"peerA": 3}, nil)
+	initial := s.GetAllPeerStats()["peerA"].RecentIncluded
+
+	// Empty block — peerA should decay.
+	s.NotifyBlock(nil, nil)
+	after := s.GetAllPeerStats()["peerA"].RecentIncluded
+
+	if after >= initial {
+		t.Fatalf("expected decay, got %f >= %f", after, initial)
+	}
+}
+
+// TestNotifyBlockDoesNotResurrectDroppedPeers verifies that finalization
+// credits to a peer with no entry don't create one.
+func TestNotifyBlockDoesNotResurrectFromFinalization(t *testing.T) {
+	s := New()
+	s.NotifyBlock(nil, map[string]int{"peerA": 5})
+
+	if stats := s.GetAllPeerStats(); len(stats) != 0 {
+		t.Fatalf("finalization credits must not create entries, got %d peers", len(stats))
+	}
+}
+
+// TestNotifyBlockDropThenFinalizeNoResurrect verifies the full drop→finalize
+// sequence: a dropped peer doesn't come back via finalization credits.
+func TestNotifyBlockDropThenFinalizeNoResurrect(t *testing.T) {
+	s := New()
+	s.NotifyBlock(map[string]int{"peerA": 1}, nil)
+	s.NotifyPeerDrop("peerA")
+	s.NotifyBlock(nil, map[string]int{"peerA": 10})
+
+	if stats := s.GetAllPeerStats(); len(stats) != 0 {
+		t.Fatalf("dropped peer must not be resurrected, got %d peers", len(stats))
+	}
+}
+
+// TestNotifyBlockFinalizationCredits an existing peer.
+func TestNotifyBlockFinalizationCredits(t *testing.T) {
+	s := New()
+	s.NotifyBlock(map[string]int{"peerA": 1}, nil)
+	s.NotifyBlock(nil, map[string]int{"peerA": 3})
+
+	// RecentFinalized is a slow EMA, not a cumulative count: assert it
+	// moved in the positive direction, not the exact value.
+	if got := s.GetAllPeerStats()["peerA"].RecentFinalized; got <= 0 {
+		t.Fatalf("expected RecentFinalized>0 after credits, got %f", got)
+	}
+}
+
+// TestNotifyBlockInclusionEMAUpdate verifies the EMA formula (1-α)·old + α·count.
+func TestNotifyBlockInclusionEMAUpdate(t *testing.T) {
+	s := New()
+	// Three inclusions: EMA = 0.05 * 3 = 0.15
+	s.NotifyBlock(map[string]int{"peerA": 3}, nil)
+	got := s.GetAllPeerStats()["peerA"].RecentIncluded
+	want := 0.15
+	if diff := got - want; diff < -1e-9 || diff > 1e-9 {
+		t.Fatalf("EMA after one sample: got %f, want %f", got, want)
+	}
+	// Next block with 10 inclusions: EMA = 0.95*0.15 + 0.05*10 = 0.6425
+	s.NotifyBlock(map[string]int{"peerA": 10}, nil)
+	got = s.GetAllPeerStats()["peerA"].RecentIncluded
+	want = 0.6425
+	if diff := got - want; diff < -1e-9 || diff > 1e-9 {
+		t.Fatalf("EMA after two samples: got %f, want %f", got, want)
+	}
+}
+
+// TestNotifyRequestResultFirstSampleBootstrap asserts that the first
+// latency sample seeds the EMA directly.
+func TestNotifyRequestResultFirstSampleBootstrap(t *testing.T) {
+	s := New()
+	s.NotifyRequestResult("peerA", 200*time.Millisecond, false)
+
+	ps := s.GetAllPeerStats()["peerA"]
+	if ps.RequestLatencyEMA != 200*time.Millisecond {
+		t.Fatalf("expected first sample to seed EMA at 200ms, got %v", ps.RequestLatencyEMA)
+	}
+	if ps.RequestSuccesses != 1 {
+		t.Fatalf("expected RequestSuccesses=1, got %d", ps.RequestSuccesses)
+	}
+	if ps.RequestTimeouts != 0 {
+		t.Fatalf("expected RequestTimeouts=0, got %d", ps.RequestTimeouts)
+	}
+}
+
+// TestNotifyRequestResultEMAUpdate verifies the EMA formula for latency.
+func TestNotifyRequestResultEMAUpdate(t *testing.T) {
+	s := New()
+	s.NotifyRequestResult("peerA", 100*time.Millisecond, false)
+	s.NotifyRequestResult("peerA", 1000*time.Millisecond, false)
+
+	// Expected: 0.99*100ms + 0.01*1000ms = 109ms
+	got := s.GetAllPeerStats()["peerA"].RequestLatencyEMA
+	want := 109 * time.Millisecond
+	delta := got - want
+	if delta < 0 {
+		delta = -delta
+	}
+	if delta > 1*time.Microsecond {
+		t.Fatalf("EMA mismatch: got %v, want %v", got, want)
+	}
+	ps := s.GetAllPeerStats()["peerA"]
+	if ps.RequestSuccesses != 2 {
+		t.Fatalf("expected RequestSuccesses=2, got %d", ps.RequestSuccesses)
+	}
+}
+
+// TestNotifyRequestResultSlowConvergence verifies the slow alpha
+// damps convergence under sustained timeouts.
+func TestNotifyRequestResultSlowConvergence(t *testing.T) {
+	s := New()
+	s.NotifyRequestResult("peerA", 100*time.Millisecond, false)
+	for i := 0; i < 50; i++ {
+		s.NotifyRequestResult("peerA", 5*time.Second, false)
+	}
+	got := s.GetAllPeerStats()["peerA"].RequestLatencyEMA
+	if got < 1*time.Second {
+		t.Fatalf("EMA did not move enough under sustained timeouts, got %v", got)
+	}
+	if got > 3*time.Second {
+		t.Fatalf("EMA converged too fast for slow alpha=0.01, got %v", got)
+	}
+}
+
+// TestNotifyPeerDropClearsStats verifies that a dropped peer disappears
+// from GetAllPeerStats.
+func TestNotifyPeerDropClearsStats(t *testing.T) {
+	s := New()
+	s.NotifyRequestResult("peerA", 200*time.Millisecond, false)
+	s.NotifyPeerDrop("peerA")
+
+	if _, ok := s.GetAllPeerStats()["peerA"]; ok {
+		t.Fatal("peerA stats should be removed after NotifyPeerDrop")
+	}
+}
+
+// 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 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)
+	s.NotifyPeerDrop("peerA")
+	// Late sample racing with the drop.
+	s.NotifyRequestResult("peerA", 50*time.Millisecond, false)
+
+	ps := s.GetAllPeerStats()["peerA"]
+	if ps.RequestSuccesses != 1 {
+		t.Fatalf("expected fresh RequestSuccesses=1, got %d", ps.RequestSuccesses)
+	}
+	if ps.RequestLatencyEMA != 50*time.Millisecond {
+		t.Fatalf("expected fresh bootstrap at 50ms, got %v", ps.RequestLatencyEMA)
+	}
+	// The dropper's MinLatencyActivity guard (in eth/dropper.go) prevents
+	// this 1-sample entry from earning latency-based protection.
+}
+
+// TestMultiplePeersIsolated verifies per-peer isolation across signal types.
+func TestMultiplePeersIsolated(t *testing.T) {
+	s := New()
+	s.NotifyBlock(map[string]int{"peerA": 5, "peerB": 0}, nil)
+	s.NotifyRequestResult("peerA", 100*time.Millisecond, false)
+	s.NotifyRequestResult("peerB", 5*time.Second, false)
+	s.NotifyBlock(nil, map[string]int{"peerA": 2})
+
+	stats := s.GetAllPeerStats()
+	// Only peerA receives finalization credits; peerB's EMA stays at zero
+	// (no credits, pure decay from zero).
+	if stats["peerA"].RecentFinalized <= 0 || stats["peerB"].RecentFinalized != 0 {
+		t.Errorf("finalization leaked: A=%f B=%f", stats["peerA"].RecentFinalized, stats["peerB"].RecentFinalized)
+	}
+	if stats["peerA"].RequestLatencyEMA != 100*time.Millisecond {
+		t.Errorf("peerA latency: got %v, want 100ms", stats["peerA"].RequestLatencyEMA)
+	}
+	if stats["peerB"].RequestLatencyEMA != 5*time.Second {
+		t.Errorf("peerB latency: got %v, want 5s", stats["peerB"].RequestLatencyEMA)
+	}
+}
+
+// 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()
+	for i := 0; i < 10; i++ {
+		s.NotifyRequestResult("peerA", 100*time.Millisecond, false)
+	}
+	s.NotifyBlock(nil, nil)
+
+	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)
+	}
+}
+
+// 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()
+	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)
+	}
+}
+
+// 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")
+	}
+}
+
+// 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)
+	}
+}
+
+// TestRequestResultTimeoutCounting verifies that timeout=true increments
+// RequestTimeouts (not RequestSuccesses) and still updates the EMA.
+func TestRequestResultTimeoutCounting(t *testing.T) {
+	s := New()
+	s.NotifyRequestResult("peerA", 5*time.Second, true)
+
+	ps := s.GetAllPeerStats()["peerA"]
+	if ps.RequestTimeouts != 1 {
+		t.Fatalf("expected RequestTimeouts=1, got %d", ps.RequestTimeouts)
+	}
+	if ps.RequestSuccesses != 0 {
+		t.Fatalf("expected RequestSuccesses=0, got %d", ps.RequestSuccesses)
+	}
+	if ps.RequestLatencyEMA != 5*time.Second {
+		t.Fatalf("EMA should bootstrap to timeout value, got %v", ps.RequestLatencyEMA)
+	}
+}
+
+// TestRequestResultMixedCounting verifies that a mix of successes and
+// timeouts increments the correct counters independently.
+func TestRequestResultMixedCounting(t *testing.T) {
+	s := New()
+	s.NotifyRequestResult("peerA", 100*time.Millisecond, false)
+	s.NotifyRequestResult("peerA", 100*time.Millisecond, false)
+	s.NotifyRequestResult("peerA", 5*time.Second, true)
+
+	ps := s.GetAllPeerStats()["peerA"]
+	if ps.RequestSuccesses != 2 {
+		t.Fatalf("expected RequestSuccesses=2, got %d", ps.RequestSuccesses)
+	}
+	if ps.RequestTimeouts != 1 {
+		t.Fatalf("expected RequestTimeouts=1, got %d", ps.RequestTimeouts)
+	}
+}

eth/txtracker/tracker.go

@@ -14,16 +14,14 @@
 // You should have received a copy of the GNU Lesser General Public License
 // along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
 
-// Package txtracker provides minimal per-peer transaction inclusion tracking.
+// Package txtracker maps accepted transactions to their delivering peer
+// and observes chain-head and finalization events to emit per-block
+// per-peer signals to a StatsConsumer (typically eth/peerstats).
 //
-// It records which peer delivered each accepted transaction (via NotifyAccepted)
-// and monitors the chain for inclusion and finalization events. When a
-// delivered transaction is finalized on chain, the delivering peer is
-// credited. A per-block exponential moving average (EMA) of inclusions
-// tracks recent peer productivity.
-//
-// The primary consumer is the peer dropper (eth/dropper.go), which uses
-// these stats to protect high-value peers from random disconnection.
+// The tracker owns the tx-hash → deliverer mapping with FIFO eviction,
+// a chain-head subscription goroutine, and the computation of per-block
+// inclusion counts and finalization credits. It does NOT maintain
+// per-peer aggregates — that is peerstats' job.
 package txtracker
 
 import (
@@ -41,21 +39,8 @@ import (
 const (
 	// Maximum number of tx→deliverer mappings to retain.
 	maxTracked = 262144
-	// EMA smoothing factor for per-block inclusion rate.
-	emaAlpha = 0.05
-	// EMA smoothing factor for per-block finalization rate. Very slow on
-	// purpose: finalization is permanent, and the score should reflect
-	// sustained contribution over long windows, not recent bursts.
-	// Half-life ≈ 6930 chain heads (~23 hours on 12s blocks).
-	finalizedEMAAlpha = 0.0001
 )
 
-// PeerStats holds the per-peer inclusion data.
-type PeerStats struct {
-	RecentFinalized float64 // EMA of per-block finalization credits (slow)
-	RecentIncluded  float64 // EMA of per-block inclusions (fast)
-}
-
 // Chain is the blockchain interface needed by the tracker.
 type Chain interface {
 	SubscribeChainHeadEvent(ch chan<- core.ChainHeadEvent) event.Subscription
@@ -64,9 +49,18 @@ type Chain interface {
 	CurrentFinalBlock() *types.Header
 }
 
-type peerStats struct {
-	recentFinalized float64
-	recentIncluded  float64
+// StatsConsumer receives per-block signals about peer inclusion and
+// finalization. The tracker invokes NotifyBlock exactly once per handled chain
+// head, AFTER releasing its own lock, with:
+//
+//   - inclusions: per-peer count of transactions in the head block
+//   - finalized:  per-peer count of transactions in blocks that became
+//     finalized since the previous call (possibly zero-range)
+//
+// Either map may be empty but the map itself is never nil when called.
+// NotifyBlock must not call back into the tracker.
+type StatsConsumer interface {
+	NotifyBlock(inclusions, finalized map[string]int)
 }
 
 // TxInfo records the per-transaction state the tracker maintains.
@@ -87,14 +81,14 @@ type TxInfo struct {
 	BlockHash common.Hash
 }
 
-// Tracker records which peer delivered each transaction and credits peers
-// when their transactions appear on chain.
+// Tracker records which peer delivered each transaction and emits
+// per-block inclusion and finalization signals to a StatsConsumer.
 type Tracker struct {
-	mu    sync.Mutex
-	txs   lru.BasicLRU[common.Hash, *TxInfo] // tx hash -> tx info with lru eviction
-	peers map[string]*peerStats
+	mu  sync.Mutex
+	txs lru.BasicLRU[common.Hash, *TxInfo] // tx hash -> tx info with lru eviction
 
 	chain        Chain
+	consumer     StatsConsumer
 	lastFinalNum uint64 // last finalized block number processed
 	headCh       chan core.ChainHeadEvent
 	sub          event.Subscription
@@ -109,17 +103,19 @@ type Tracker struct {
 // New creates a new tracker.
 func New() *Tracker {
 	return &Tracker{
-		txs:   lru.NewBasicLRU[common.Hash, *TxInfo](maxTracked),
-		peers: make(map[string]*peerStats),
-		quit:  make(chan struct{}),
-		step:  make(chan struct{}, 1),
-		now:   func() uint64 { return uint64(time.Now().Unix()) },
+		txs:  lru.NewBasicLRU[common.Hash, *TxInfo](maxTracked),
+		quit: make(chan struct{}),
+		step: make(chan struct{}, 1),
+		now:  func() uint64 { return uint64(time.Now().Unix()) },
 	}
 }
 
-// Start begins listening for chain head events.
-func (t *Tracker) Start(chain Chain) {
+// Start begins listening for chain head events. `consumer` receives
+// per-block signals; if nil, signals are computed but discarded
+// (useful in tests that exercise only the tx-lifecycle surface).
+func (t *Tracker) Start(chain Chain, consumer StatsConsumer) {
 	t.chain = chain
+	t.consumer = consumer
 	// Seed lastFinalNum so checkFinalization doesn't backfill from genesis.
 	if fh := chain.CurrentFinalBlock(); fh != nil {
 		t.lastFinalNum = fh.Number.Uint64()
@@ -130,14 +126,6 @@ func (t *Tracker) Start(chain Chain) {
 	go t.loop()
 }
 
-// NotifyPeerDrop removes a disconnected peer's stats to prevent unbounded
-// growth. Safe to call from any goroutine.
-func (t *Tracker) NotifyPeerDrop(peer string) {
-	t.mu.Lock()
-	defer t.mu.Unlock()
-	delete(t.peers, peer)
-}
-
 // Stop shuts down the tracker.
 func (t *Tracker) Stop() {
 	t.stopOnce.Do(func() {
@@ -164,26 +152,6 @@ func (t *Tracker) NotifyAccepted(peer string, hashes []common.Hash) {
 		}
 		t.txs.Add(hash, &TxInfo{Deliverer: peer, AddedAt: addedAt})
 	}
-	// Ensure the delivering peer has a stats entry.
-	if len(hashes) > 0 && t.peers[peer] == nil {
-		t.peers[peer] = &peerStats{}
-	}
-}
-
-// GetAllPeerStats returns a snapshot of per-peer inclusion statistics.
-// Safe to call from any goroutine.
-func (t *Tracker) GetAllPeerStats() map[string]PeerStats {
-	t.mu.Lock()
-	defer t.mu.Unlock()
-
-	result := make(map[string]PeerStats, len(t.peers))
-	for id, ps := range t.peers {
-		result[id] = PeerStats{
-			RecentFinalized: ps.recentFinalized,
-			RecentIncluded:  ps.recentIncluded,
-		}
-	}
-	return result
 }
 
 func (t *Tracker) loop() {
@@ -205,6 +173,10 @@ func (t *Tracker) loop() {
 	}
 }
 
+// handleChainHead computes per-peer deltas for the new head block and any
+// newly-finalized blocks, then hands them to the StatsConsumer AFTER
+// releasing t.mu. The lock-release-before-consumer pattern avoids any
+// cross-package lock ordering.
 func (t *Tracker) handleChainHead(ev core.ChainHeadEvent) {
 	// Fetch the head block by hash (not just number) to avoid using a
 	// reorged block if the tracker goroutine lags behind the chain.
@@ -213,39 +185,35 @@ func (t *Tracker) handleChainHead(ev core.ChainHeadEvent) {
 		return
 	}
 	t.mu.Lock()
-	defer t.mu.Unlock()
 
-	// Count per-peer inclusions in this block for the inclusion EMA, and
-	// record (BlockNum, BlockHash) on first inclusion so the iterate-t.txs
-	// finalization scan can find the entry later without re-reading the
-	// block. Skip txs whose delivery arrived at or after this block's slot
-	// — those are likely post-slot re-broadcasts of an already-mined tx,
-	// not genuine relay work.
+	// Count per-peer inclusions in this block, and record (BlockNum,
+	// BlockHash) on first inclusion so the iterate-t.txs finalization
+	// scan can find the entry later without re-reading the block. Skip
+	// txs whose delivery arrived at or after this block's slot — those
+	// are likely post-slot re-broadcasts of an already-mined tx, not
+	// genuine relay work.
 	blockTime := block.Time()
 	blockNum := block.Number().Uint64()
 	blockHash := block.Hash()
-	blockIncl := make(map[string]int)
+	inclusions := make(map[string]int)
 	for _, tx := range block.Transactions() {
 		ti, ok := t.txs.Peek(tx.Hash())
 		if !ok || ti.AddedAt >= blockTime {
 			continue
 		}
-		blockIncl[ti.Deliverer]++
+		inclusions[ti.Deliverer]++
 		if ti.BlockNum == 0 {
 			ti.BlockNum = blockNum
 			ti.BlockHash = blockHash
 		}
 	}
 	// Accumulate per-peer finalization credits over the newly-finalized
-	// range (possibly zero blocks). Only counts peers still tracked.
-	blockFinal := t.collectFinalizationCredits()
+	// range (possibly zero blocks).
+	finalized := t.collectFinalizationCredits()
+	t.mu.Unlock()
 
-	// Update both EMAs for all tracked peers (decays inactive ones).
-	// Don't create entries for unknown peers — they may have been
-	// removed by NotifyPeerDrop and should not be resurrected.
-	for peer, ps := range t.peers {
-		ps.recentIncluded = (1-emaAlpha)*ps.recentIncluded + emaAlpha*float64(blockIncl[peer])
-		ps.recentFinalized = (1-finalizedEMAAlpha)*ps.recentFinalized + finalizedEMAAlpha*float64(blockFinal[peer])
+	if t.consumer != nil {
+		t.consumer.NotifyBlock(inclusions, finalized)
 	}
 }
 

eth/txtracker/tracker_test.go

@@ -75,8 +75,7 @@ func (c *mockChain) CurrentFinalBlock() *types.Header {
 	return &types.Header{Number: new(big.Int).SetUint64(c.finalNum)}
 }
 
-// addBlock adds a canonical block at the given height. Overwrites any
-// prior canonical block at that height.
+// addBlock adds a canonical block at the given height.
 func (c *mockChain) addBlock(num uint64, txs []*types.Transaction) *types.Block {
 	return c.addBlockAtHeight(num, num, txs, true)
 }
@@ -95,7 +94,6 @@ func (c *mockChain) addBlockAtHeight(num, salt uint64, txs []*types.Transaction,
 func (c *mockChain) addBlockAtHeightWithTime(num, salt uint64, txs []*types.Transaction, canonical bool, blockTime uint64) *types.Block {
 	c.mu.Lock()
 	defer c.mu.Unlock()
-	// Mix salt into Extra so siblings at the same height get distinct hashes.
 	header := &types.Header{
 		Number: new(big.Int).SetUint64(num),
 		Extra:  big.NewInt(int64(salt)).Bytes(),
@@ -115,9 +113,7 @@ func (c *mockChain) setFinalBlock(num uint64) {
 	c.finalNum = num
 }
 
-// sendHead emits a chain head event for the canonical block at the given
-// height. The emitted header carries the real block's hash so the
-// tracker's GetBlock(hash, number) lookup resolves correctly.
+// sendHead emits a chain head event for the canonical block at the given height.
 func (c *mockChain) sendHead(num uint64) {
 	c.mu.Lock()
 	hash := c.canonicalByNum[num]
@@ -147,6 +143,43 @@ func makeTx(nonce uint64) *types.Transaction {
 	return types.NewTx(&types.LegacyTx{Nonce: nonce, GasPrice: big.NewInt(1), Gas: 21000})
 }
 
+// mockConsumer captures NotifyBlock invocations so tests can assert on the
+// signals the tracker emits.
+type mockConsumer struct {
+	mu      sync.Mutex
+	signals []signal
+}
+
+type signal struct {
+	inclusions, finalized map[string]int
+}
+
+func (c *mockConsumer) NotifyBlock(inclusions, finalized map[string]int) {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	// Deep-copy so tests inspecting older signals aren't tripped up by
+	// later iterations mutating the same map (they don't today, but
+	// this keeps the assertion model simple).
+	in := make(map[string]int, len(inclusions))
+	for k, v := range inclusions {
+		in[k] = v
+	}
+	fn := make(map[string]int, len(finalized))
+	for k, v := range finalized {
+		fn[k] = v
+	}
+	c.signals = append(c.signals, signal{in, fn})
+}
+
+func (c *mockConsumer) last() signal {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	if len(c.signals) == 0 {
+		return signal{}
+	}
+	return c.signals[len(c.signals)-1]
+}
+
 // waitStep blocks until the tracker has processed one event.
 func waitStep(t *testing.T, tr *Tracker) {
 	t.Helper()
@@ -157,232 +190,187 @@ func waitStep(t *testing.T, tr *Tracker) {
 	}
 }
 
-func TestNotifyReceived(t *testing.T) {
+// TestNotifyAcceptedRecordsMapping verifies the tx-lifecycle surface:
+// NotifyAccepted records tx→peer mappings in insertion order, with
+// first-deliverer-wins semantics on duplicates.
+func TestNotifyAcceptedRecordsMapping(t *testing.T) {
 	tr := New()
-	chain := newMockChain()
-	tr.Start(chain)
-	defer tr.Stop()
 
 	txs := []*types.Transaction{makeTx(1), makeTx(2), makeTx(3)}
 	hashes := hashTxs(txs)
 	tr.NotifyAccepted("peerA", hashes)
 
-	// Public surface: peer entry was created with zero stats before any
-	// chain events. Map lookups would return a zero value for a missing
-	// key, so assert presence explicitly.
-	stats := tr.GetAllPeerStats()
-	if len(stats) != 1 {
-		t.Fatalf("expected 1 peer entry, got %d", len(stats))
-	}
-	ps, ok := stats["peerA"]
-	if !ok {
-		t.Fatal("expected peerA entry, not found")
-	}
-	if ps.RecentFinalized != 0 || ps.RecentIncluded != 0 {
-		t.Fatalf("expected zero stats before chain events, got %+v", ps)
-	}
-
-	// Internal state: all tx→deliverer mappings recorded.
 	tr.mu.Lock()
 	defer tr.mu.Unlock()
 	if tr.txs.Len() != 3 {
 		t.Fatalf("expected 3 tracked txs, got %d", tr.txs.Len())
 	}
+	// Keys() walks the internal list from the least-recently-added end,
+	// which for the tracker's add-once/Peek-only usage is insertion order.
+	for i, h := range tr.txs.Keys() {
+		if hashes[i] != h {
+			t.Fatalf("insertion order mismatch at %d", i)
+		}
+	}
 	for i, h := range hashes {
-		got, ok := tr.txs.Peek(h)
+		ti, ok := tr.txs.Peek(h)
 		if !ok {
 			t.Fatalf("tx %d: not tracked", i)
 		}
-		if got.Deliverer != "peerA" {
-			t.Fatalf("tx %d: expected deliverer=peerA, got %q", i, got.Deliverer)
+		if ti.Deliverer != "peerA" {
+			t.Fatalf("tx %d: expected deliverer=peerA, got %q", i, ti.Deliverer)
 		}
 	}
 }
 
-func TestInclusionEMA(t *testing.T) {
+// TestNotifyAcceptedFirstDelivererWins verifies duplicate accepts
+// preserve the original deliverer.
+func TestNotifyAcceptedFirstDelivererWins(t *testing.T) {
 	tr := New()
-	chain := newMockChain()
-	tr.Start(chain)
-	defer tr.Stop()
-
 	tx := makeTx(1)
 	tr.NotifyAccepted("peerA", []common.Hash{tx.Hash()})
+	tr.NotifyAccepted("peerB", []common.Hash{tx.Hash()})
 
-	// Block 1 includes peerA's tx.
-	chain.addBlock(1, []*types.Transaction{tx})
-	chain.sendHead(1)
-	waitStep(t, tr)
-
-	stats := tr.GetAllPeerStats()
-	if stats["peerA"].RecentIncluded <= 0 {
-		t.Fatalf("expected RecentIncluded > 0 after inclusion, got %f", stats["peerA"].RecentIncluded)
+	tr.mu.Lock()
+	defer tr.mu.Unlock()
+	ti, ok := tr.txs.Peek(tx.Hash())
+	if !ok {
+		t.Fatal("tx not tracked")
 	}
-	ema1 := stats["peerA"].RecentIncluded
-
-	// Block 2 has no txs from peerA — EMA should decay.
-	chain.addBlock(2, nil)
-	chain.sendHead(2)
-	waitStep(t, tr)
-
-	stats = tr.GetAllPeerStats()
-	if stats["peerA"].RecentIncluded >= ema1 {
-		t.Fatalf("expected EMA to decay, got %f >= %f", stats["peerA"].RecentIncluded, ema1)
+	if ti.Deliverer != "peerA" {
+		t.Fatalf("expected first deliverer peerA to win, got %q", ti.Deliverer)
+	}
+	if tr.txs.Len() != 1 {
+		t.Fatalf("expected single tracked tx, got %d", tr.txs.Len())
 	}
 }
 
-func TestFinalization(t *testing.T) {
+// TestHandleChainHeadEmitsInclusions verifies the tracker emits a
+// correct per-peer inclusion map to its consumer when a head block
+// contains tracked transactions.
+func TestHandleChainHeadEmitsInclusions(t *testing.T) {
 	tr := New()
 	chain := newMockChain()
-	tr.Start(chain)
+	consumer := &mockConsumer{}
+	tr.Start(chain, consumer)
 	defer tr.Stop()
 
-	tx := makeTx(1)
-	tr.NotifyAccepted("peerA", []common.Hash{tx.Hash()})
-
-	// Include in block 1.
-	chain.addBlock(1, []*types.Transaction{tx})
-	chain.sendHead(1)
-	waitStep(t, tr)
-
-	// Not finalized yet.
-	stats := tr.GetAllPeerStats()
-	if stats["peerA"].RecentFinalized != 0 {
-		t.Fatalf("expected RecentFinalized=0 before finalization, got %f", stats["peerA"].RecentFinalized)
-	}
-
-	// Finalize block 1, then send head 2 to trigger the finalization EMA update.
-	chain.setFinalBlock(1)
-	chain.addBlock(2, nil)
-	chain.sendHead(2)
-	waitStep(t, tr)
-
-	stats = tr.GetAllPeerStats()
-	if stats["peerA"].RecentFinalized <= 0 {
-		t.Fatalf("expected RecentFinalized>0 after finalization, got %f", stats["peerA"].RecentFinalized)
-	}
-}
-
-func TestMultiplePeers(t *testing.T) {
-	tr := New()
-	chain := newMockChain()
-	tr.Start(chain)
-	defer tr.Stop()
-
-	tx1 := makeTx(1)
-	tx2 := makeTx(2)
+	tx1, tx2 := makeTx(1), makeTx(2)
 	tr.NotifyAccepted("peerA", []common.Hash{tx1.Hash()})
 	tr.NotifyAccepted("peerB", []common.Hash{tx2.Hash()})
 
-	// Both included in block 1.
 	chain.addBlock(1, []*types.Transaction{tx1, tx2})
 	chain.sendHead(1)
 	waitStep(t, tr)
 
-	// Finalize.
+	sig := consumer.last()
+	if sig.inclusions["peerA"] != 1 {
+		t.Errorf("peerA inclusions: got %d, want 1", sig.inclusions["peerA"])
+	}
+	if sig.inclusions["peerB"] != 1 {
+		t.Errorf("peerB inclusions: got %d, want 1", sig.inclusions["peerB"])
+	}
+	if len(sig.finalized) != 0 {
+		t.Errorf("expected empty finalized map, got %v", sig.finalized)
+	}
+}
+
+// TestHandleChainHeadEmptyBlock verifies an empty head block emits an
+// empty inclusion map (so peerstats can decay all known peers).
+func TestHandleChainHeadEmptyBlock(t *testing.T) {
+	tr := New()
+	chain := newMockChain()
+	consumer := &mockConsumer{}
+	tr.Start(chain, consumer)
+	defer tr.Stop()
+
+	chain.addBlock(1, nil)
+	chain.sendHead(1)
+	waitStep(t, tr)
+
+	sig := consumer.last()
+	if len(sig.inclusions) != 0 {
+		t.Errorf("expected empty inclusions, got %v", sig.inclusions)
+	}
+}
+
+// TestHandleChainHeadEmitsFinalization verifies that when finalization
+// advances, the consumer receives per-peer finalization credits
+// accumulated over the newly-finalized range.
+func TestHandleChainHeadEmitsFinalization(t *testing.T) {
+	tr := New()
+	chain := newMockChain()
+	consumer := &mockConsumer{}
+	tr.Start(chain, consumer)
+	defer tr.Stop()
+
+	tx := makeTx(1)
+	tr.NotifyAccepted("peerA", []common.Hash{tx.Hash()})
+
+	// Include in block 1, not yet finalized.
+	chain.addBlock(1, []*types.Transaction{tx})
+	chain.sendHead(1)
+	waitStep(t, tr)
+
+	if credits := consumer.last().finalized["peerA"]; credits != 0 {
+		t.Fatalf("expected no finalization credits before finalization, got %d", credits)
+	}
+
+	// Finalize block 1; next head triggers the finalization scan.
 	chain.setFinalBlock(1)
 	chain.addBlock(2, nil)
 	chain.sendHead(2)
 	waitStep(t, tr)
 
-	stats := tr.GetAllPeerStats()
-	if stats["peerA"].RecentFinalized <= 0 {
-		t.Fatalf("peerA: expected RecentFinalized>0, got %f", stats["peerA"].RecentFinalized)
-	}
-	if stats["peerB"].RecentFinalized <= 0 {
-		t.Fatalf("peerB: expected RecentFinalized>0, got %f", stats["peerB"].RecentFinalized)
+	if credits := consumer.last().finalized["peerA"]; credits != 1 {
+		t.Fatalf("expected 1 finalization credit, got %d", credits)
 	}
 }
 
-func TestFirstDelivererWins(t *testing.T) {
+// TestFinalizationSkipsOrphanedInclusion verifies the finalization scan
+// re-checks the recorded inclusion block hash against the canonical chain:
+// a tx whose recorded inclusion block was reorged out must not yield a
+// finalization credit.
+func TestFinalizationSkipsOrphanedInclusion(t *testing.T) {
 	tr := New()
 	chain := newMockChain()
-	tr.Start(chain)
+	consumer := &mockConsumer{}
+	tr.Start(chain, consumer)
 	defer tr.Stop()
 
 	tx := makeTx(1)
 	tr.NotifyAccepted("peerA", []common.Hash{tx.Hash()})
-	tr.NotifyAccepted("peerB", []common.Hash{tx.Hash()}) // duplicate, should be ignored
 
-	chain.addBlock(1, []*types.Transaction{tx})
-	chain.sendHead(1)
+	// The tx is first seen in block A at height 1 (canonical at the time),
+	// recording (BlockNum=1, BlockHash=A).
+	blockA := chain.addBlockAtHeight(1, 1, []*types.Transaction{tx}, true)
+	chain.sendHeadBlock(blockA)
 	waitStep(t, tr)
 
+	// Reorg: sibling B (without the tx) becomes canonical at height 1.
+	chain.addBlockAtHeight(1, 2, nil, true)
+
+	// Finalize height 1; next head triggers the finalization scan. The
+	// recorded hash A no longer matches canonical-at-1 (now B), so no
+	// credit may be emitted.
 	chain.setFinalBlock(1)
 	chain.addBlock(2, nil)
 	chain.sendHead(2)
 	waitStep(t, tr)
 
-	stats := tr.GetAllPeerStats()
-	if stats["peerA"].RecentFinalized <= 0 {
-		t.Fatalf("peerA should be credited, got RecentFinalized=%f", stats["peerA"].RecentFinalized)
-	}
-	if stats["peerB"].RecentFinalized != 0 {
-		t.Fatalf("peerB should NOT be credited, got RecentFinalized=%f", stats["peerB"].RecentFinalized)
+	if credits := consumer.last().finalized["peerA"]; credits != 0 {
+		t.Fatalf("expected no credit for orphaned inclusion, got %d", credits)
 	}
 }
 
-func TestNoFinalizationCredit(t *testing.T) {
-	tr := New()
-	chain := newMockChain()
-	tr.Start(chain)
-	defer tr.Stop()
-
-	tx := makeTx(1)
-	tr.NotifyAccepted("peerA", []common.Hash{tx.Hash()})
-
-	// Include but don't finalize.
-	chain.addBlock(1, []*types.Transaction{tx})
-	chain.sendHead(1)
-	waitStep(t, tr)
-
-	// Send more heads without finalization.
-	chain.addBlock(2, nil)
-	chain.sendHead(2)
-	waitStep(t, tr)
-
-	stats := tr.GetAllPeerStats()
-	if stats["peerA"].RecentFinalized != 0 {
-		t.Fatalf("expected RecentFinalized=0 without finalization, got %f", stats["peerA"].RecentFinalized)
-	}
-}
-
-func TestEMADecay(t *testing.T) {
-	tr := New()
-	chain := newMockChain()
-	tr.Start(chain)
-	defer tr.Stop()
-
-	tx := makeTx(1)
-	tr.NotifyAccepted("peerA", []common.Hash{tx.Hash()})
-
-	// Include in block 1.
-	chain.addBlock(1, []*types.Transaction{tx})
-	chain.sendHead(1)
-	waitStep(t, tr)
-
-	// Send 30 empty blocks — EMA should decay close to zero.
-	for i := uint64(2); i <= 31; i++ {
-		chain.addBlock(i, nil)
-		chain.sendHead(i)
-		waitStep(t, tr)
-	}
-
-	stats := tr.GetAllPeerStats()
-	if stats["peerA"].RecentIncluded > 0.02 {
-		t.Fatalf("expected RecentIncluded near zero after 30 empty blocks, got %f", stats["peerA"].RecentIncluded)
-	}
-}
-
-// TestReorgSafety verifies that handleChainHead resolves the head block by
-// HASH (not just by number), so a head event announcing a sibling block at
-// the same height does not credit transactions from the canonical block.
-//
-// Regression check: handleChainHead uses GetBlock(hash, number) so a head
-// event announcing sibling B fetches B, not the canonical block A.
+// TestReorgSafety verifies the tracker resolves the head block by HASH
+// so a head event pointing at a sibling block does not emit inclusions
+// from the canonical block at the same height.
 func TestReorgSafety(t *testing.T) {
 	tr := New()
 	chain := newMockChain()
-	tr.Start(chain)
+	consumer := &mockConsumer{}
+	tr.Start(chain, consumer)
 	defer tr.Stop()
 
 	tx := makeTx(1)
@@ -395,77 +383,33 @@ func TestReorgSafety(t *testing.T) {
 		t.Fatal("sibling blocks ended up with the same hash")
 	}
 
-	// Head announces sibling B. A hash-aware tracker fetches B, sees no
-	// peerA txs, and leaves the EMA at zero. A number-only tracker would
-	// instead fetch A and credit peerA.
+	// Head announces sibling B — emit must contain no peerA inclusions.
 	chain.sendHeadBlock(blockB)
 	waitStep(t, tr)
-
-	if got := tr.GetAllPeerStats()["peerA"].RecentIncluded; got != 0 {
-		t.Fatalf("expected RecentIncluded=0 after sibling-B head event, got %f (tracker followed the wrong block)", got)
+	if incl := consumer.last().inclusions["peerA"]; incl != 0 {
+		t.Fatalf("sibling-B head should emit 0 peerA inclusions, got %d", incl)
 	}
 
-	// Now announce canonical A; peerA should be credited.
+	// Head announces canonical A — emit must contain 1 peerA inclusion.
 	chain.sendHeadBlock(blockA)
 	waitStep(t, tr)
-
-	if got := tr.GetAllPeerStats()["peerA"].RecentIncluded; got <= 0 {
-		t.Fatalf("expected RecentIncluded>0 after canonical-A head event, got %f", got)
-	}
-}
-
-// TestRecentFinalizedDecays verifies that the finalization EMA decays
-// for a peer that earned credits in the past but has no new
-// finalization activity. The decay is slow (α=0.0001), so we
-// just assert monotonic decrease, not convergence to zero.
-func TestRecentFinalizedDecays(t *testing.T) {
-	tr := New()
-	chain := newMockChain()
-	tr.Start(chain)
-	defer tr.Stop()
-
-	tx := makeTx(1)
-	tr.NotifyAccepted("peerA", []common.Hash{tx.Hash()})
-
-	// Include and finalize in block 1.
-	chain.addBlock(1, []*types.Transaction{tx})
-	chain.sendHead(1)
-	waitStep(t, tr)
-	chain.setFinalBlock(1)
-	chain.addBlock(2, nil)
-	chain.sendHead(2)
-	waitStep(t, tr)
-
-	peak := tr.GetAllPeerStats()["peerA"].RecentFinalized
-	if peak <= 0 {
-		t.Fatalf("expected RecentFinalized>0 after finalization, got %f", peak)
-	}
-
-	// Send many empty heads — peer contributes zero each block,
-	// EMA should decay monotonically.
-	for i := uint64(3); i <= 50; i++ {
-		chain.addBlock(i, nil)
-		chain.sendHead(i)
-		waitStep(t, tr)
-	}
-
-	after := tr.GetAllPeerStats()["peerA"].RecentFinalized
-	if after >= peak {
-		t.Fatalf("expected RecentFinalized to decay, got %f >= peak %f", after, peak)
+	if incl := consumer.last().inclusions["peerA"]; incl != 1 {
+		t.Fatalf("canonical-A head should emit 1 peerA inclusion, got %d", incl)
 	}
 }
 
 // TestPreSlotGate verifies that a tx delivered at or after the slot of its
-// inclusion block earns no credit. This blocks the simple
-// post-block-propagation re-broadcast attribution attack: a peer that
+// inclusion block is not reported in the inclusion signal. This blocks the
+// simple post-block-propagation re-broadcast attribution attack: a peer that
 // learns a tx from the just-mined block and re-broadcasts it to our pool
 // should not gain credit when that block is processed. The finalization
-// path applies the same gate (ti.AddedAt >= blockTime) and is exercised
-// by the existing TestFinalization with the new clock semantics.
+// path applies the same gate (ti.AddedAt >= blockTime) because entries
+// skipped here never record a BlockNum/BlockHash.
 func TestPreSlotGate(t *testing.T) {
 	tr := New()
 	chain := newMockChain()
-	tr.Start(chain)
+	consumer := &mockConsumer{}
+	tr.Start(chain, consumer)
 	defer tr.Stop()
 
 	// Pin the tracker's clock so NotifyAccepted records a known addedAt.
@@ -482,9 +426,8 @@ func TestPreSlotGate(t *testing.T) {
 	chain.sendHead(1)
 	waitStep(t, tr)
 
-	preEMA := tr.GetAllPeerStats()["peerA"].RecentIncluded
-	if preEMA <= 0 {
-		t.Fatalf("expected RecentIncluded>0 after pre-slot delivery, got %f", preEMA)
+	if incl := consumer.last().inclusions["peerA"]; incl != 1 {
+		t.Fatalf("expected 1 inclusion for pre-slot delivery, got %d", incl)
 	}
 
 	// Block 2: slot strictly BEFORE delivery — post-slot, must NOT credit.
@@ -492,10 +435,20 @@ func TestPreSlotGate(t *testing.T) {
 	chain.sendHead(2)
 	waitStep(t, tr)
 
-	// With the gate, only EMA decay occurs (no contribution this block).
-	// Without the gate, RecentIncluded would have ticked up again.
-	after := tr.GetAllPeerStats()["peerA"].RecentIncluded
-	if after >= preEMA {
-		t.Fatalf("expected EMA to decay (no credit for post-slot tx), got %f >= preEMA %f", after, preEMA)
+	if incl := consumer.last().inclusions["peerA"]; incl != 0 {
+		t.Fatalf("expected 0 inclusions for post-slot delivery, got %d", incl)
 	}
 }
+
+// TestHandleChainHeadNilConsumer verifies the tracker tolerates a nil
+// consumer (useful for tests that only exercise tx-lifecycle behavior).
+func TestHandleChainHeadNilConsumer(t *testing.T) {
+	tr := New()
+	chain := newMockChain()
+	tr.Start(chain, nil)
+	defer tr.Stop()
+
+	chain.addBlock(1, nil)
+	chain.sendHead(1)
+	waitStep(t, tr) // should not panic
+}

tests/fuzzers/txfetcher/txfetcher_fuzzer.go

@@ -92,6 +92,7 @@ func fuzz(input []byte) int {
 		func(string, []common.Hash) error { return nil },
 		nil,
 		nil,
+		nil,
 		blobpool.NewBlobBuffer(blobpool.BlobBufferFunctions{
 			ValidateTx: func(*types.Transaction) error { return nil },
 			AddToPool:  func(*blobpool.BlobTxForPool) error { return nil },