use std::collections::HashMap;
use std::time::Instant;
#[derive(Debug, Clone)]
pub struct ScoreParameter {
pub topic_weight: f64,
pub delivery_weight: f64,
pub behaviour_weight: f64,
pub decay_interval_secs: u64,
pub score_threshold_ban: f64,
pub score_threshold_graylist: f64,
}
impl Default for ScoreParameter {
fn default() -> Self {
Self {
topic_weight: 0.5,
delivery_weight: 0.3,
behaviour_weight: 0.2,
decay_interval_secs: 10,
score_threshold_ban: -100.0,
score_threshold_graylist: -10.0,
}
}
}
#[derive(Debug, Clone, Default)]
pub struct TopicScore {
pub first_message_deliveries: f64,
pub mesh_message_deliveries: f64,
pub mesh_failure_penalty: f64,
pub invalid_message_deliveries: f64,
pub in_mesh: bool,
pub mesh_time_secs: f64,
}
impl TopicScore {
pub fn score(&self, p: &ScoreParameter) -> f64 {
let p1 = self.first_message_deliveries;
let p2 = self.mesh_message_deliveries;
let p3 = -self.mesh_failure_penalty;
let p4 = -self.invalid_message_deliveries * 10.0;
(p1 + p2 + p3 + p4) * p.topic_weight
}
}
#[derive(Debug, Clone, PartialEq)]
pub enum BehaviourPenalty {
InvalidMessage,
GraftBackoff,
PromiscuousPX,
AppSpecific(f64),
}
impl BehaviourPenalty {
pub fn delta(&self) -> f64 {
match self {
BehaviourPenalty::InvalidMessage => -10.0,
BehaviourPenalty::GraftBackoff => -5.0,
BehaviourPenalty::PromiscuousPX => -3.0,
BehaviourPenalty::AppSpecific(v) => *v,
}
}
}
#[derive(Debug, Clone)]
pub struct PeerScore {
pub peer_id: String,
pub topic_scores: HashMap<String, TopicScore>,
pub behaviour_penalties: f64,
pub app_specific_score: f64,
pub last_decay: Instant,
}
impl PeerScore {
pub fn new(peer_id: impl Into<String>) -> Self {
Self {
peer_id: peer_id.into(),
topic_scores: HashMap::new(),
behaviour_penalties: 0.0,
app_specific_score: 0.0,
last_decay: Instant::now(),
}
}
pub fn composite_score(&self, params: &ScoreParameter) -> f64 {
let topic_sum: f64 = self.topic_scores.values().map(|ts| ts.score(params)).sum();
let behaviour_part = self.behaviour_penalties * params.behaviour_weight;
let app_part = self.app_specific_score * params.delivery_weight;
topic_sum + behaviour_part + app_part
}
pub fn apply_penalty(&mut self, p: BehaviourPenalty) {
self.behaviour_penalties += p.delta();
}
pub fn record_first_delivery(&mut self, topic: &str) {
let ts = self.topic_scores.entry(topic.to_string()).or_default();
ts.first_message_deliveries += 1.0;
}
pub fn record_mesh_delivery(&mut self, topic: &str) {
let ts = self.topic_scores.entry(topic.to_string()).or_default();
ts.mesh_message_deliveries += 1.0;
}
pub fn record_invalid_message(&mut self, topic: &str) {
let ts = self.topic_scores.entry(topic.to_string()).or_default();
ts.invalid_message_deliveries += 1.0;
}
pub fn set_in_mesh(&mut self, topic: &str, in_mesh: bool) {
let ts = self.topic_scores.entry(topic.to_string()).or_default();
ts.in_mesh = in_mesh;
}
pub fn decay(&mut self, params: &ScoreParameter) {
let elapsed = self.last_decay.elapsed();
let factor = 0.9_f64.powf(elapsed.as_secs_f64() / params.decay_interval_secs as f64);
for ts in self.topic_scores.values_mut() {
ts.first_message_deliveries *= factor;
ts.mesh_message_deliveries *= factor;
ts.mesh_failure_penalty *= factor;
ts.invalid_message_deliveries *= factor;
ts.mesh_time_secs *= factor;
}
self.behaviour_penalties *= factor;
self.app_specific_score *= factor;
self.last_decay = Instant::now();
}
}
#[derive(Debug)]
pub struct PeerScoreTracker {
pub scores: HashMap<String, PeerScore>,
pub params: ScoreParameter,
}
impl PeerScoreTracker {
pub fn new(params: ScoreParameter) -> Self {
Self {
scores: HashMap::new(),
params,
}
}
pub fn get_or_create(&mut self, peer_id: &str) -> &mut PeerScore {
self.scores
.entry(peer_id.to_string())
.or_insert_with(|| PeerScore::new(peer_id))
}
pub fn composite_score(&self, peer_id: &str) -> f64 {
self.scores
.get(peer_id)
.map(|ps| ps.composite_score(&self.params))
.unwrap_or(0.0)
}
pub fn banned_peers(&self) -> Vec<String> {
self.scores
.iter()
.filter(|(_, ps)| ps.composite_score(&self.params) < self.params.score_threshold_ban)
.map(|(id, _)| id.clone())
.collect()
}
pub fn greylisted_peers(&self) -> Vec<String> {
self.scores
.iter()
.filter(|(_, ps)| {
let s = ps.composite_score(&self.params);
s < self.params.score_threshold_graylist && s >= self.params.score_threshold_ban
})
.map(|(id, _)| id.clone())
.collect()
}
pub fn best_peers(&self, n: usize) -> Vec<String> {
let mut scored: Vec<(String, f64)> = self
.scores
.iter()
.map(|(id, ps)| (id.clone(), ps.composite_score(&self.params)))
.collect();
scored.sort_by(|a, b| b.1.partial_cmp(&a.1).unwrap_or(std::cmp::Ordering::Equal));
scored.into_iter().take(n).map(|(id, _)| id).collect()
}
pub fn apply_decay_all(&mut self) {
let params = self.params.clone();
for ps in self.scores.values_mut() {
ps.decay(¶ms);
}
}
pub fn prune_peers(&mut self, ids: &[&str]) {
for id in ids {
self.scores.remove(*id);
}
}
}
#[cfg(test)]
mod tests {
use super::*;
use std::time::Duration;
#[test]
fn test_default_params() {
let p = ScoreParameter::default();
assert!((p.topic_weight - 0.5).abs() < f64::EPSILON);
assert!((p.delivery_weight - 0.3).abs() < f64::EPSILON);
assert!((p.behaviour_weight - 0.2).abs() < f64::EPSILON);
assert_eq!(p.decay_interval_secs, 10);
assert!((p.score_threshold_ban - (-100.0)).abs() < f64::EPSILON);
assert!((p.score_threshold_graylist - (-10.0)).abs() < f64::EPSILON);
}
#[test]
fn test_topic_score_formula() {
let p = ScoreParameter::default();
let ts = TopicScore {
first_message_deliveries: 4.0,
mesh_message_deliveries: 6.0,
mesh_failure_penalty: 1.0,
invalid_message_deliveries: 0.5,
in_mesh: true,
mesh_time_secs: 30.0,
};
let expected = (4.0 + 6.0 - 1.0 - 5.0) * 0.5;
assert!((ts.score(&p) - expected).abs() < 1e-9);
}
#[test]
fn test_behaviour_penalty_deltas() {
assert!((BehaviourPenalty::InvalidMessage.delta() - (-10.0)).abs() < f64::EPSILON);
assert!((BehaviourPenalty::GraftBackoff.delta() - (-5.0)).abs() < f64::EPSILON);
assert!((BehaviourPenalty::PromiscuousPX.delta() - (-3.0)).abs() < f64::EPSILON);
assert!((BehaviourPenalty::AppSpecific(7.5).delta() - 7.5).abs() < f64::EPSILON);
assert!((BehaviourPenalty::AppSpecific(-2.0).delta() - (-2.0)).abs() < f64::EPSILON);
}
#[test]
fn test_composite_score_weighted_sum() {
let p = ScoreParameter::default();
let mut ps = PeerScore::new("peer-a");
ps.app_specific_score = 10.0;
ps.behaviour_penalties = -5.0;
let c = ps.composite_score(&p);
assert!((c - 2.0).abs() < 1e-9, "got {c}");
}
#[test]
fn test_record_first_delivery() {
let mut ps = PeerScore::new("peer-b");
ps.record_first_delivery("topic-1");
ps.record_first_delivery("topic-1");
let ts = ps.topic_scores.get("topic-1").expect("topic must exist");
assert!((ts.first_message_deliveries - 2.0).abs() < f64::EPSILON);
assert!((ts.mesh_message_deliveries).abs() < f64::EPSILON);
}
#[test]
fn test_record_mesh_delivery() {
let mut ps = PeerScore::new("peer-c");
ps.record_mesh_delivery("topic-2");
ps.record_mesh_delivery("topic-2");
ps.record_mesh_delivery("topic-2");
let ts = ps.topic_scores.get("topic-2").expect("topic must exist");
assert!((ts.mesh_message_deliveries - 3.0).abs() < f64::EPSILON);
assert!((ts.first_message_deliveries).abs() < f64::EPSILON);
}
#[test]
fn test_record_invalid_message_penalises() {
let p = ScoreParameter::default();
let mut ps = PeerScore::new("peer-d");
let before = ps.composite_score(&p);
ps.record_invalid_message("topic-3");
let after = ps.composite_score(&p);
assert!(after < before, "invalid message must lower composite score");
let ts = ps.topic_scores.get("topic-3").expect("topic must exist");
assert!((ts.invalid_message_deliveries - 1.0).abs() < f64::EPSILON);
}
#[test]
fn test_set_in_mesh() {
let mut ps = PeerScore::new("peer-e");
ps.set_in_mesh("t", true);
assert!(ps.topic_scores["t"].in_mesh);
ps.set_in_mesh("t", false);
assert!(!ps.topic_scores["t"].in_mesh);
}
#[test]
fn test_apply_penalty_variants() {
let mut ps = PeerScore::new("peer-f");
assert!((ps.behaviour_penalties).abs() < f64::EPSILON);
ps.apply_penalty(BehaviourPenalty::InvalidMessage);
assert!((ps.behaviour_penalties - (-10.0)).abs() < f64::EPSILON);
ps.apply_penalty(BehaviourPenalty::GraftBackoff);
assert!((ps.behaviour_penalties - (-15.0)).abs() < f64::EPSILON);
ps.apply_penalty(BehaviourPenalty::PromiscuousPX);
assert!((ps.behaviour_penalties - (-18.0)).abs() < f64::EPSILON);
ps.apply_penalty(BehaviourPenalty::AppSpecific(3.0));
assert!((ps.behaviour_penalties - (-15.0)).abs() < f64::EPSILON);
}
#[test]
fn test_decay_reduces_fields() {
let p = ScoreParameter {
decay_interval_secs: 1,
..ScoreParameter::default()
};
let mut ps = PeerScore::new("peer-g");
ps.record_first_delivery("t");
ps.record_mesh_delivery("t");
ps.record_invalid_message("t");
ps.behaviour_penalties = -20.0;
ps.app_specific_score = 5.0;
ps.last_decay = Instant::now() - Duration::from_secs(10);
ps.decay(&p);
let ts = &ps.topic_scores["t"];
assert!(ts.first_message_deliveries < 1.0);
assert!(ts.mesh_message_deliveries < 1.0);
assert!(ts.invalid_message_deliveries < 1.0);
assert!(ps.behaviour_penalties > -20.0);
assert!(ps.app_specific_score < 5.0);
}
#[test]
fn test_banned_peers() {
let p = ScoreParameter::default();
let mut tracker = PeerScoreTracker::new(p);
{
let ps = tracker.get_or_create("good");
ps.app_specific_score = 10.0;
}
{
let ps = tracker.get_or_create("bad");
ps.behaviour_penalties = -600.0;
}
let banned = tracker.banned_peers();
assert!(banned.contains(&"bad".to_string()));
assert!(!banned.contains(&"good".to_string()));
}
#[test]
fn test_greylisted_peers() {
let p = ScoreParameter::default();
let mut tracker = PeerScoreTracker::new(p);
{
let ps = tracker.get_or_create("ok");
ps.app_specific_score = 5.0;
}
{
let ps = tracker.get_or_create("grey");
ps.behaviour_penalties = -75.0;
}
{
let ps = tracker.get_or_create("banned");
ps.behaviour_penalties = -600.0;
}
let grey = tracker.greylisted_peers();
assert!(grey.contains(&"grey".to_string()));
assert!(!grey.contains(&"ok".to_string()));
assert!(!grey.contains(&"banned".to_string()));
}
#[test]
fn test_best_peers_sorted() {
let p = ScoreParameter::default();
let mut tracker = PeerScoreTracker::new(p);
for (id, score) in [("p1", 5.0), ("p2", 20.0), ("p3", 1.0), ("p4", 10.0)] {
let ps = tracker.get_or_create(id);
ps.app_specific_score = score;
}
let best = tracker.best_peers(3);
assert_eq!(best.len(), 3);
assert_eq!(best[0], "p2");
assert_eq!(best[1], "p4");
assert_eq!(best[2], "p1");
}
#[test]
fn test_prune_peers() {
let p = ScoreParameter::default();
let mut tracker = PeerScoreTracker::new(p);
tracker.get_or_create("a");
tracker.get_or_create("b");
tracker.get_or_create("c");
assert_eq!(tracker.scores.len(), 3);
tracker.prune_peers(&["a", "c"]);
assert_eq!(tracker.scores.len(), 1);
assert!(tracker.scores.contains_key("b"));
assert!(!tracker.scores.contains_key("a"));
assert!(!tracker.scores.contains_key("c"));
}
#[test]
fn test_get_or_create() {
let p = ScoreParameter::default();
let mut tracker = PeerScoreTracker::new(p);
assert!(!tracker.scores.contains_key("new-peer"));
tracker.get_or_create("new-peer");
assert!(tracker.scores.contains_key("new-peer"));
tracker.get_or_create("new-peer");
assert_eq!(tracker.scores.len(), 1);
}
#[test]
fn test_apply_decay_all() {
let p = ScoreParameter {
decay_interval_secs: 1,
..ScoreParameter::default()
};
let mut tracker = PeerScoreTracker::new(p);
for id in ["x", "y", "z"] {
let ps = tracker.get_or_create(id);
ps.behaviour_penalties = -50.0;
ps.last_decay = Instant::now() - Duration::from_secs(10);
}
tracker.apply_decay_all();
for id in ["x", "y", "z"] {
let ps = &tracker.scores[id];
assert!(ps.behaviour_penalties > -50.0, "decay must reduce {id}");
assert!(ps.behaviour_penalties < 0.0, "{id} still negative");
}
}
#[test]
fn test_composite_score_unknown_peer() {
let tracker = PeerScoreTracker::new(ScoreParameter::default());
assert!((tracker.composite_score("ghost")).abs() < f64::EPSILON);
}
}