diff --git a/eth/backend.go b/eth/backend.go index ba9b4d4459..5b9721ad60 100644 --- a/eth/backend.go +++ b/eth/backend.go @@ -479,7 +479,7 @@ func (s *Ethereum) Start() error { s.handler.Start(s.p2pServer.MaxPeers) // Start the connection manager with inclusion-based peer protection. - s.dropper.Start(s.p2pServer, func() bool { return !s.Synced() }, s.handler.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) diff --git a/eth/dropper.go b/eth/dropper.go index 36c970e8d8..53b425e3ac 100644 --- a/eth/dropper.go +++ b/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() diff --git a/eth/dropper_test.go b/eth/dropper_test.go index fd2ed9b611..92556f4250 100644 --- a/eth/dropper_test.go +++ b/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") + } +} diff --git a/eth/fetcher/tx_fetcher.go b/eth/fetcher/tx_fetcher.go index 63efdfab60..cd0ea065b3 100644 --- a/eth/fetcher/tx_fetcher.go +++ b/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) diff --git a/eth/fetcher/tx_fetcher_test.go b/eth/fetcher/tx_fetcher_test.go index 1c13c5bfc7..37d401779a 100644 --- a/eth/fetcher/tx_fetcher_test.go +++ b/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) + } + }), + }, + }) +} diff --git a/eth/handler.go b/eth/handler.go index ed0149075f..ae72b856d1 100644 --- a/eth/handler.go +++ b/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) diff --git a/eth/peerstats/peerstats.go b/eth/peerstats/peerstats.go new file mode 100644 index 0000000000..7ab8422781 --- /dev/null +++ b/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 . + +// 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 +} diff --git a/eth/peerstats/peerstats_test.go b/eth/peerstats/peerstats_test.go new file mode 100644 index 0000000000..a3fa8bdeb0 --- /dev/null +++ b/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 . + +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) + } +} diff --git a/eth/txtracker/tracker.go b/eth/txtracker/tracker.go index 3ddd0231a8..db81b3b4d1 100644 --- a/eth/txtracker/tracker.go +++ b/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 . -// 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) } } diff --git a/eth/txtracker/tracker_test.go b/eth/txtracker/tracker_test.go index 8e7e81d034..e179d4440f 100644 --- a/eth/txtracker/tracker_test.go +++ b/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 +} diff --git a/tests/fuzzers/txfetcher/txfetcher_fuzzer.go b/tests/fuzzers/txfetcher/txfetcher_fuzzer.go index 6e9b8acd83..ad4a2ff30d 100644 --- a/tests/fuzzers/txfetcher/txfetcher_fuzzer.go +++ b/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 },