diff --git a/polkadot/node/network/collator-protocol/src/validator_side_experimental/peer_manager/db.rs b/polkadot/node/network/collator-protocol/src/validator_side_experimental/peer_manager/db.rs index 4fbb40da32e8..7ee6732552a7 100644 --- a/polkadot/node/network/collator-protocol/src/validator_side_experimental/peer_manager/db.rs +++ b/polkadot/node/network/collator-protocol/src/validator_side_experimental/peer_manager/db.rs @@ -21,6 +21,7 @@ use crate::validator_side_experimental::{ use async_trait::async_trait; use polkadot_node_network_protocol::PeerId; use polkadot_primitives::{BlockNumber, Id as ParaId}; +use sp_runtime::{traits::Bounded, FixedPointNumber, FixedU128}; use std::{ collections::{btree_map, hash_map, BTreeMap, BTreeSet, HashMap}, time::{SystemTime, UNIX_EPOCH}, @@ -186,9 +187,14 @@ impl Db { per_para_entry.remove(); } else if per_para_entry.get().len() > per_para_limit { // We have exceeded the maximum capacity, in which case we need to prune - // the least recently bumped values let diff = per_para_entry.get().len() - per_para_limit; - Self::prune_for_para(¶, &mut per_para_entry, diff, &mut reported_updates); + Self::prune_for_para( + ¶, + &mut per_para_entry, + diff, + now, + &mut reported_updates, + ); } } } @@ -196,17 +202,29 @@ impl Db { reported_updates } + // Evicts the entries with minimum `score / (age in milliseconds)` ratio. fn prune_for_para( para_id: &ParaId, per_para: &mut btree_map::OccupiedEntry>, diff: usize, + now: Timestamp, reported_updates: &mut Vec, ) { for _ in 0..diff { let (peer_id_to_remove, score) = per_para .get() .iter() - .min_by_key(|(_peer, entry)| entry.last_bumped) + .min_by_key(|(_peer, entry)| { + let age = now.saturating_sub(entry.last_bumped); + let score = u16::from(entry.score); + let ratio = FixedU128::checked_from_rational(u128::from(score), age) + .unwrap_or(FixedU128::max_value()); + // In case of equal ratios, we evict the entry with the lower absolute score. + // Note: Multiple peers can have the exact same (ratio, score) if they were + // updated in the same batch (sharing the same `last_bumped` timestamp) and + // have identical scores. In such cases, the eviction choice is arbitrary. + (ratio, score) + }) .map(|(peer, entry)| (*peer, entry.score)) .expect("We know there are enough reps over the limit"); @@ -276,7 +294,7 @@ mod tests { assert_eq!(db.processed_finalized_block_number().await, Some(10)); assert_eq!(db.len(), 0); - // Test a query on a non-existant entry. + // Test a query on a non-existent entry. assert_eq!(db.query(&PeerId::random(), &ParaId::from(1000)).await, None); // Test empty update with decay. @@ -340,7 +358,7 @@ mod tests { assert_eq!(db.processed_finalized_block_number().await, Some(13)); assert_eq!(db.len(), 1); assert_eq!(db.query(&first_peer_id, &first_para_id).await.unwrap(), Score::new(10)); - // Query a non-existant peer_id for this para. + // Query a non-existent peer_id for this para. assert_eq!(db.query(&PeerId::random(), &first_para_id).await, None); // Query this peer's rep for a different para. assert_eq!(db.query(&first_peer_id, &ParaId::from(200)).await, None); @@ -697,4 +715,295 @@ mod tests { assert_eq!(db.len(), 0); assert_eq!(db.query(&peer_id, &ParaId::from(300)).await, None); } + + mod peer_pruning { + use super::*; + + #[tokio::test] + async fn max_score_oldest_first() { + use crate::validator_side_experimental::common::MAX_SCORE; + use std::time::{SystemTime, UNIX_EPOCH}; + + let mut db = Db::new(2).await; + let para_id = ParaId::from(100); + let peer_old = PeerId::random(); + let peer_mid = PeerId::random(); + let peer_new = PeerId::random(); + + let now = SystemTime::now().duration_since(UNIX_EPOCH).unwrap().as_millis(); + + // Inject two existing peers with MAX_SCORE two days apart; peer_old has a lower + // score/age ratio (older with equal score) and will be evicted. + let mut reputations = HashMap::new(); + reputations.insert( + peer_old, + ScoreEntry { + score: Score::new(MAX_SCORE), + last_bumped: now.saturating_sub(172_800_000), + }, + ); + reputations.insert( + peer_mid, + ScoreEntry { + score: Score::new(MAX_SCORE), + last_bumped: now.saturating_sub(86_400_000), + }, + ); + db.set_para_reputations(para_id, reputations); + + // Adding peer_new pushes the para over the limit of 2, triggering one eviction. + db.process_bumps( + 1, + [(para_id, [(peer_new, Score::new(MAX_SCORE))].into_iter().collect())] + .into_iter() + .collect(), + None, + ) + .await; + + // The oldest peer (smallest last_bumped) should have been evicted. + assert_eq!(db.query(&peer_old, ¶_id).await, None, "oldest peer should be pruned"); + assert!(db.query(&peer_mid, ¶_id).await.is_some(), "middle peer should remain"); + assert!(db.query(&peer_new, ¶_id).await.is_some(), "newest peer should remain"); + } + + #[tokio::test] + async fn lower_score_over_older_timestamp() { + use crate::validator_side_experimental::common::MAX_SCORE; + use std::time::{SystemTime, UNIX_EPOCH}; + + let mut db = Db::new(2).await; + let para_id = ParaId::from(200); + let now = SystemTime::now().duration_since(UNIX_EPOCH).unwrap().as_millis(); + let mut reputations = HashMap::new(); + + // score=32767, age=100s -> ratio≈0.328 + let peer_high_score = PeerId::random(); + reputations.insert( + peer_high_score, + ScoreEntry { + score: Score::new(MAX_SCORE / 2), + last_bumped: now.saturating_sub(100_000), + }, + ); + + // score=100, age=1s -> ratio=0.1 + let peer_low_score = PeerId::random(); + reputations.insert( + peer_low_score, + ScoreEntry { score: Score::new(100), last_bumped: now.saturating_sub(1_000) }, + ); + db.set_para_reputations(para_id, reputations); + + // score=1, age=0 -> ratio=max_value + let peer_trigger = PeerId::random(); + db.process_bumps( + 1, + [(para_id, [(peer_trigger, Score::new(1))].into_iter().collect())] + .into_iter() + .collect(), + None, + ) + .await; + + assert!(db.query(&peer_high_score, ¶_id).await.is_some(), "ratio≈0.328, survives"); + assert_eq!(db.query(&peer_low_score, ¶_id).await, None, "ratio=0.1, evicted"); + } + + // A high-score peer remains in the DB even as low-score peers cycle through it. + // When a fresh low-score peer is added and the limit is exceeded, the oldest + // low-score peer (lowest score/age ratio) is evicted rather than the high-score one. + #[tokio::test] + async fn high_score_peer_protected_from_low_score_churn() { + use std::time::{SystemTime, UNIX_EPOCH}; + + let mut db = Db::new(4).await; + let para_id = ParaId::from(100); + let now = SystemTime::now().duration_since(UNIX_EPOCH).unwrap().as_millis(); + let mut reputations = HashMap::new(); + + let peer_high = PeerId::random(); // score=2000, age=5s -> ratio=0.4 + reputations.insert( + peer_high, + ScoreEntry { score: Score::new(2000), last_bumped: now.saturating_sub(5_000) }, + ); + + // score=1, age=30s -> ratio=1/30_000 + let peer_old = PeerId::random(); + reputations.insert( + peer_old, + ScoreEntry { score: Score::new(1), last_bumped: now.saturating_sub(30_000) }, + ); + + // score=1, age=20s -> ratio=1/20_000 + let peer_mid = PeerId::random(); + reputations.insert( + peer_mid, + ScoreEntry { score: Score::new(1), last_bumped: now.saturating_sub(20_000) }, + ); + + // score=1, age=10s -> ratio=1/10_000 + let peer_recent = PeerId::random(); + reputations.insert( + peer_recent, + ScoreEntry { score: Score::new(1), last_bumped: now.saturating_sub(10_000) }, + ); + db.set_para_reputations(para_id, reputations); + + // peer_new pushes over the limit of 4, triggering one eviction + // score=1, age=0 -> ratio=max_value + let peer_new = PeerId::random(); + db.process_bumps( + 1, + [(para_id, [(peer_new, Score::new(1))].into_iter().collect())] + .into_iter() + .collect(), + None, + ) + .await; + + assert_eq!(db.query(&peer_old, ¶_id).await, None, "ratio=1/30_000, evicted"); + assert!(db.query(&peer_high, ¶_id).await.is_some(), "ratio=0.4, survives"); + assert!(db.query(&peer_mid, ¶_id).await.is_some()); + assert!(db.query(&peer_recent, ¶_id).await.is_some()); + assert!(db.query(&peer_new, ¶_id).await.is_some()); + } + + #[tokio::test] + async fn multiple_evictions_correct_order() { + use std::time::{SystemTime, UNIX_EPOCH}; + + let mut db = Db::new(2).await; // 5 entries → 3 evictions needed + let para_id = ParaId::from(100); + let now = SystemTime::now().duration_since(UNIX_EPOCH).unwrap().as_millis(); + let mut reputations = HashMap::new(); + + // score=1, age=100s -> ratio=1/100_000 + let peer_a = PeerId::random(); + reputations.insert( + peer_a, + ScoreEntry { score: Score::new(1), last_bumped: now.saturating_sub(100_000) }, + ); + + // score=1, age=50s -> ratio=1/50_000 + let peer_b = PeerId::random(); + reputations.insert( + peer_b, + ScoreEntry { score: Score::new(1), last_bumped: now.saturating_sub(50_000) }, + ); + + // score=1, age=20s -> ratio=1/20_000 + let peer_c = PeerId::random(); + reputations.insert( + peer_c, + ScoreEntry { score: Score::new(1), last_bumped: now.saturating_sub(20_000) }, + ); + + // score=2000, age=1000s -> ratio=0.002 + let peer_d = PeerId::random(); + reputations.insert( + peer_d, + ScoreEntry { score: Score::new(2000), last_bumped: now.saturating_sub(1_000_000) }, + ); + db.set_para_reputations(para_id, reputations); + + // score=1, age=0 -> ratio=max_value + let peer_e = PeerId::random(); + db.process_bumps( + // peer_e triggers 3 evictions (5 entries → limit 2) + 1, + [(para_id, [(peer_e, Score::new(1))].into_iter().collect())] + .into_iter() + .collect(), + None, + ) + .await; + + assert_eq!(db.query(&peer_a, ¶_id).await, None, "ratio=1/100_000, evicted"); + assert_eq!(db.query(&peer_b, ¶_id).await, None, "ratio=1/50_000, evicted"); + assert_eq!(db.query(&peer_c, ¶_id).await, None, "ratio=1/20_000, evicted"); + assert!(db.query(&peer_d, ¶_id).await.is_some(), "ratio=0.002, survives"); + assert!(db.query(&peer_e, ¶_id).await.is_some(), "ratio=max_value, survives"); + } + + #[tokio::test] + async fn zero_score_works() { + use std::time::{SystemTime, UNIX_EPOCH}; + + let mut db = Db::new(2).await; + let para_id = ParaId::from(100); + let now = SystemTime::now().duration_since(UNIX_EPOCH).unwrap().as_millis(); + let mut reputations = HashMap::new(); + + // score=0, age=1s → checked_from_rational(0, age) = Some(0) → ratio=0 + let peer_zero = PeerId::random(); + reputations.insert( + peer_zero, + ScoreEntry { score: Score::new(0), last_bumped: now.saturating_sub(1_000) }, + ); + + // score=100, age=10s → ratio=0.01 + let peer_normal = PeerId::random(); + reputations.insert( + peer_normal, + ScoreEntry { score: Score::new(100), last_bumped: now.saturating_sub(10_000) }, + ); + db.set_para_reputations(para_id, reputations); + + let peer_trigger = PeerId::random(); + db.process_bumps( + 1, + [(para_id, [(peer_trigger, Score::new(1))].into_iter().collect())] + .into_iter() + .collect(), + None, + ) + .await; + + assert_eq!(db.query(&peer_zero, ¶_id).await, None, "ratio=0, evicted"); + assert!(db.query(&peer_normal, ¶_id).await.is_some(), "ratio=0.01, survives"); + assert!(db.query(&peer_trigger, ¶_id).await.is_some(), "ratio=max_value, survives"); + } + + #[tokio::test] + async fn equal_ratio_tiebreaker_evicts_lower_score() { + use std::time::{SystemTime, UNIX_EPOCH}; + + // Both entries have the same score/age ratio (2/200 = 3/300 = 0.01). The tiebreaker + // `score` breaks the tie deterministically: lower score → evicted first. + let mut db = Db::new(2).await; + let para_id = ParaId::from(100); + let now = SystemTime::now().duration_since(UNIX_EPOCH).unwrap().as_millis(); + let mut reputations = HashMap::new(); + + // score=2, age=200ms -> ratio=0.01, tiebreaker=2 + let peer_a = PeerId::random(); + reputations.insert( + peer_a, + ScoreEntry { score: Score::new(2), last_bumped: now.saturating_sub(200) }, + ); + + // score=3, age=300ms -> ratio=0.01, tiebreaker=3 + let peer_b = PeerId::random(); + reputations.insert( + peer_b, + ScoreEntry { score: Score::new(3), last_bumped: now.saturating_sub(300) }, + ); + db.set_para_reputations(para_id, reputations); + + let peer_trigger = PeerId::random(); + db.process_bumps( + 1, + [(para_id, [(peer_trigger, Score::new(1))].into_iter().collect())] + .into_iter() + .collect(), + None, + ) + .await; + + assert_eq!(db.query(&peer_a, ¶_id).await, None, "lower score wins tie, evicted"); + assert!(db.query(&peer_b, ¶_id).await.is_some(), "peer_b survives"); + assert!(db.query(&peer_trigger, ¶_id).await.is_some(), "peer_trigger survives"); + } + } } diff --git a/prdoc/pr_11576.prdoc b/prdoc/pr_11576.prdoc new file mode 100644 index 000000000000..3d710b711d4d --- /dev/null +++ b/prdoc/pr_11576.prdoc @@ -0,0 +1,13 @@ +title: 'collator-protocol revamp: use ratio for peer eviction from DB ' +doc: +- audience: Node Dev + description: |- + The current implementation of `prune_for_para` gives edge to new peers because it uses only the + timestamp of the last bump when evicting peers from the DB. As a result, a high score collator + which is inactive for a while can easily be evicted by new peers with minimal score. + + To fix this we now calculate `score / time_since_last_bump` ratio for each peer and evict the one + with the min value. +crates: +- name: polkadot-collator-protocol + bump: patch