use crate::PeerId;
use parking_lot::RwLock;
use serde::{Deserialize, Serialize};
use std::collections::HashMap;
use std::sync::Arc;
use std::time::{Instant, SystemTime, UNIX_EPOCH};
pub const DEFAULT_NEUTRAL_TRUST: f64 = 0.5;
const MIN_TRUST_SCORE: f64 = 0.0;
const MAX_TRUST_SCORE: f64 = 1.0;
const EMA_WEIGHT: f64 = 0.1;
const DECAY_LAMBDA: f64 = 1.3761e-6;
#[derive(Debug, Clone)]
struct PeerTrust {
score: f64,
last_updated: Instant,
}
impl PeerTrust {
fn new() -> Self {
Self {
score: DEFAULT_NEUTRAL_TRUST,
last_updated: Instant::now(),
}
}
fn apply_decay(&mut self) {
let elapsed_secs = self.last_updated.elapsed().as_secs_f64();
self.apply_decay_secs(elapsed_secs);
}
fn apply_decay_secs(&mut self, elapsed_secs: f64) {
if elapsed_secs > 0.0 {
let decay_factor = (-DECAY_LAMBDA * elapsed_secs).exp();
self.score =
DEFAULT_NEUTRAL_TRUST + (self.score - DEFAULT_NEUTRAL_TRUST) * decay_factor;
self.score = self.score.clamp(MIN_TRUST_SCORE, MAX_TRUST_SCORE);
self.last_updated = Instant::now();
}
}
fn record_weighted(&mut self, observation: f64, weight: f64) {
if !weight.is_finite() || weight <= 0.0 {
return;
}
self.apply_decay();
let alpha_w = 1.0 - (1.0 - EMA_WEIGHT).powf(weight);
self.score = (1.0 - alpha_w) * self.score + alpha_w * observation;
self.score = self.score.clamp(MIN_TRUST_SCORE, MAX_TRUST_SCORE);
self.last_updated = Instant::now();
}
#[allow(dead_code)] fn record(&mut self, observation: f64) {
self.record_weighted(observation, 1.0);
}
fn decayed_score(&self) -> f64 {
Self::decay_score(self.score, self.last_updated.elapsed().as_secs_f64())
}
fn decay_score(score: f64, elapsed_secs: f64) -> f64 {
if elapsed_secs > 0.0 {
let decay_factor = (-DECAY_LAMBDA * elapsed_secs).exp();
let decayed = DEFAULT_NEUTRAL_TRUST + (score - DEFAULT_NEUTRAL_TRUST) * decay_factor;
decayed.clamp(MIN_TRUST_SCORE, MAX_TRUST_SCORE)
} else {
score
}
}
}
const SUCCESS_OBSERVATION: f64 = 1.0;
const FAILURE_OBSERVATION: f64 = 0.0;
#[derive(Debug, Clone)]
pub enum NodeStatisticsUpdate {
CorrectResponse,
FailedResponse,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct TrustSnapshot {
pub peers: HashMap<PeerId, TrustRecord>,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct TrustRecord {
pub score: f64,
pub last_updated_epoch_secs: u64,
}
#[derive(Debug)]
pub struct TrustEngine {
peers: Arc<RwLock<HashMap<PeerId, PeerTrust>>>,
}
impl TrustEngine {
pub fn new() -> Self {
Self {
peers: Arc::new(RwLock::new(HashMap::new())),
}
}
pub fn update_node_stats(&self, node_id: &PeerId, update: NodeStatisticsUpdate) {
self.update_node_stats_weighted(node_id, update, 1.0);
}
pub fn update_node_stats_weighted(
&self,
node_id: &PeerId,
update: NodeStatisticsUpdate,
weight: f64,
) {
let mut peers = self.peers.write();
let entry = peers.entry(*node_id).or_insert_with(PeerTrust::new);
let observation = match update {
NodeStatisticsUpdate::CorrectResponse => SUCCESS_OBSERVATION,
NodeStatisticsUpdate::FailedResponse => FAILURE_OBSERVATION,
};
entry.record_weighted(observation, weight);
}
pub fn score(&self, node_id: &PeerId) -> f64 {
let peers = self.peers.read();
peers
.get(node_id)
.map(|p| p.decayed_score())
.unwrap_or(DEFAULT_NEUTRAL_TRUST)
}
pub fn remove_node(&self, node_id: &PeerId) {
let mut peers = self.peers.write();
peers.remove(node_id);
}
pub fn export_snapshot(&self) -> TrustSnapshot {
let peers_guard = self.peers.read();
let now_epoch = SystemTime::now()
.duration_since(UNIX_EPOCH)
.map(|d| d.as_secs())
.unwrap_or(0);
let peers = peers_guard
.iter()
.map(|(peer_id, peer_trust)| {
let record = TrustRecord {
score: peer_trust.decayed_score(),
last_updated_epoch_secs: now_epoch,
};
(*peer_id, record)
})
.collect();
TrustSnapshot { peers }
}
pub fn import_snapshot(&self, snapshot: &TrustSnapshot) {
let mut peers_guard = self.peers.write();
for (peer_id, record) in &snapshot.peers {
let score = if record.score.is_finite() {
record.score.clamp(MIN_TRUST_SCORE, MAX_TRUST_SCORE)
} else {
DEFAULT_NEUTRAL_TRUST
};
let peer_trust = PeerTrust {
score,
last_updated: Instant::now(),
};
peers_guard.insert(*peer_id, peer_trust);
}
}
#[cfg(test)]
pub async fn simulate_elapsed(&self, node_id: &PeerId, elapsed: std::time::Duration) {
let mut peers = self.peers.write();
if let Some(trust) = peers.get_mut(node_id) {
trust.apply_decay_secs(elapsed.as_secs_f64());
}
}
}
impl Default for TrustEngine {
fn default() -> Self {
Self::new()
}
}
#[cfg(test)]
mod tests {
use super::*;
#[tokio::test]
async fn test_unknown_peer_returns_neutral() {
let engine = TrustEngine::new();
let peer = PeerId::random();
assert!((engine.score(&peer) - DEFAULT_NEUTRAL_TRUST).abs() < f64::EPSILON);
}
#[tokio::test]
async fn test_successes_increase_score() {
let engine = TrustEngine::new();
let peer = PeerId::random();
for _ in 0..50 {
engine.update_node_stats(&peer, NodeStatisticsUpdate::CorrectResponse);
}
let score = engine.score(&peer);
assert!(
score > DEFAULT_NEUTRAL_TRUST,
"Score {score} should be above neutral"
);
assert!(score <= MAX_TRUST_SCORE, "Score {score} should be <= max");
}
#[tokio::test]
async fn test_failures_decrease_score() {
let engine = TrustEngine::new();
let peer = PeerId::random();
for _ in 0..50 {
engine.update_node_stats(&peer, NodeStatisticsUpdate::FailedResponse);
}
let score = engine.score(&peer);
assert!(
score < DEFAULT_NEUTRAL_TRUST,
"Score {score} should be below neutral"
);
assert!(score >= MIN_TRUST_SCORE, "Score {score} should be >= min");
}
#[tokio::test]
async fn test_scores_clamped_to_bounds() {
let engine = TrustEngine::new();
let peer = PeerId::random();
for _ in 0..1000 {
engine.update_node_stats(&peer, NodeStatisticsUpdate::CorrectResponse);
}
let score = engine.score(&peer);
assert!(score >= MIN_TRUST_SCORE, "Score {score} below min");
assert!(score <= MAX_TRUST_SCORE, "Score {score} above max");
for _ in 0..2000 {
engine.update_node_stats(&peer, NodeStatisticsUpdate::FailedResponse);
}
let score = engine.score(&peer);
assert!(score >= MIN_TRUST_SCORE, "Score {score} below min");
assert!(score <= MAX_TRUST_SCORE, "Score {score} above max");
}
#[tokio::test]
async fn test_remove_node_resets_to_neutral() {
let engine = TrustEngine::new();
let peer = PeerId::random();
engine.update_node_stats(&peer, NodeStatisticsUpdate::FailedResponse);
assert!(engine.score(&peer) < DEFAULT_NEUTRAL_TRUST);
engine.remove_node(&peer);
assert!((engine.score(&peer) - DEFAULT_NEUTRAL_TRUST).abs() < f64::EPSILON);
}
#[tokio::test]
async fn test_ema_blends_observations() {
let engine = TrustEngine::new();
let peer = PeerId::random();
engine.update_node_stats(&peer, NodeStatisticsUpdate::FailedResponse);
let after_fail = engine.score(&peer);
assert!(after_fail < DEFAULT_NEUTRAL_TRUST);
engine.update_node_stats(&peer, NodeStatisticsUpdate::CorrectResponse);
let after_success = engine.score(&peer);
assert!(after_success > after_fail, "Success should increase score");
}
#[test]
fn test_worst_score_unblocks_after_3_days() {
let three_days_secs = (3 * 24 * 3600) as f64;
let score = PeerTrust::decay_score(MIN_TRUST_SCORE, three_days_secs);
assert!(
score >= 0.15,
"After 3 days, score {score} should be >= block threshold 0.15",
);
}
#[test]
fn test_worst_score_still_blocked_before_3_days() {
let just_under_3_days = (3 * 24 * 3600 - 3600) as f64; let score = PeerTrust::decay_score(MIN_TRUST_SCORE, just_under_3_days);
assert!(
score < 0.15,
"Before 3 days, score {score} should still be < block threshold 0.15",
);
}
#[test]
fn test_decay_from_high_score_moves_down() {
let one_week_secs = (7 * 24 * 3600) as f64;
let score = PeerTrust::decay_score(0.95, one_week_secs);
assert!(score < 0.95, "Score should have decayed from 0.95");
assert!(
score > DEFAULT_NEUTRAL_TRUST,
"Score should still be above neutral after 1 week"
);
}
#[test]
fn test_decay_from_low_score_moves_up() {
let one_week_secs = (7 * 24 * 3600) as f64;
let score = PeerTrust::decay_score(0.1, one_week_secs);
assert!(score > 0.1, "Low score should decay upward toward neutral");
}
#[tokio::test]
async fn test_export_import_roundtrip() {
let engine = TrustEngine::new();
let peer1 = PeerId::random();
let peer2 = PeerId::random();
for _ in 0..20 {
engine.update_node_stats(&peer1, NodeStatisticsUpdate::CorrectResponse);
}
for _ in 0..10 {
engine.update_node_stats(&peer2, NodeStatisticsUpdate::FailedResponse);
}
let score1_before = engine.score(&peer1);
let score2_before = engine.score(&peer2);
let snapshot = engine.export_snapshot();
assert_eq!(snapshot.peers.len(), 2);
let engine2 = TrustEngine::new();
engine2.import_snapshot(&snapshot);
let score1_after = engine2.score(&peer1);
let score2_after = engine2.score(&peer2);
assert!(
(score1_before - score1_after).abs() < 0.01,
"peer1 score drifted: before={score1_before}, after={score1_after}"
);
assert!(
(score2_before - score2_after).abs() < 0.01,
"peer2 score drifted: before={score2_before}, after={score2_after}"
);
}
#[tokio::test]
async fn test_import_preserves_scores_without_decay() {
let peer = PeerId::random();
let one_day_secs: u64 = 86_400;
let one_day_ago = SystemTime::now()
.duration_since(UNIX_EPOCH)
.unwrap()
.as_secs()
- one_day_secs;
let snapshot = TrustSnapshot {
peers: HashMap::from([(
peer,
TrustRecord {
score: 0.9,
last_updated_epoch_secs: one_day_ago,
},
)]),
};
let engine = TrustEngine::new();
engine.import_snapshot(&snapshot);
let score = engine.score(&peer);
assert!(
(score - 0.9).abs() < 0.01,
"Score {score} should be ~0.9 (no offline decay)"
);
}
#[tokio::test]
async fn test_import_nan_score_falls_back_to_neutral() {
let peer = PeerId::random();
let snapshot = TrustSnapshot {
peers: HashMap::from([(
peer,
TrustRecord {
score: f64::NAN,
last_updated_epoch_secs: 1_000_000,
},
)]),
};
let engine = TrustEngine::new();
engine.import_snapshot(&snapshot);
let score = engine.score(&peer);
assert!(
score.is_finite(),
"NaN score should have been replaced with a finite value"
);
assert!(
(score - DEFAULT_NEUTRAL_TRUST).abs() < f64::EPSILON,
"NaN score should fall back to neutral, got {score}"
);
}
#[tokio::test]
async fn test_import_infinity_score_falls_back_to_neutral() {
let peer = PeerId::random();
let snapshot = TrustSnapshot {
peers: HashMap::from([(
peer,
TrustRecord {
score: f64::INFINITY,
last_updated_epoch_secs: 1_000_000,
},
)]),
};
let engine = TrustEngine::new();
engine.import_snapshot(&snapshot);
let score = engine.score(&peer);
assert!(
score.is_finite(),
"Infinity score should have been replaced with a finite value"
);
assert!(
(score - DEFAULT_NEUTRAL_TRUST).abs() < f64::EPSILON,
"Infinity score should fall back to neutral, got {score}"
);
}
#[tokio::test]
async fn test_negative_weight_is_noop() {
let engine = TrustEngine::new();
let peer = PeerId::random();
let before = engine.score(&peer);
engine.update_node_stats_weighted(&peer, NodeStatisticsUpdate::FailedResponse, -5.0);
let after_negative = engine.score(&peer);
assert!(
(before - after_negative).abs() < f64::EPSILON,
"negative weight should be a no-op: before={before}, after={after_negative}"
);
engine.update_node_stats_weighted(&peer, NodeStatisticsUpdate::CorrectResponse, -1.0);
let after_negative_success = engine.score(&peer);
assert!(
(before - after_negative_success).abs() < f64::EPSILON,
"negative weight success should be a no-op: before={before}, after={after_negative_success}"
);
engine.update_node_stats_weighted(&peer, NodeStatisticsUpdate::FailedResponse, 1.0);
let after_valid = engine.score(&peer);
assert!(
after_valid < before,
"valid weight-1 failure should reduce score: before={before}, after={after_valid}"
);
}
#[tokio::test]
async fn test_weighted_ema_larger_impact() {
let engine = TrustEngine::new();
let peer_a = PeerId::random();
let peer_b = PeerId::random();
engine.update_node_stats_weighted(&peer_a, NodeStatisticsUpdate::FailedResponse, 1.0);
let score_a = engine.score(&peer_a);
engine.update_node_stats_weighted(&peer_b, NodeStatisticsUpdate::FailedResponse, 5.0);
let score_b = engine.score(&peer_b);
assert!(
score_b < score_a,
"weight-5 failure ({score_b}) should produce lower score than weight-1 ({score_a})"
);
}
#[tokio::test]
async fn test_unit_weight_equivalence() {
let engine1 = TrustEngine::new();
let engine2 = TrustEngine::new();
let peer = PeerId::random();
engine1.update_node_stats(&peer, NodeStatisticsUpdate::FailedResponse);
engine2.update_node_stats_weighted(&peer, NodeStatisticsUpdate::FailedResponse, 1.0);
let diff = (engine1.score(&peer) - engine2.score(&peer)).abs();
assert!(
diff < 1e-10,
"unit-weight paths should be equivalent, diff={diff}"
);
}
#[tokio::test]
async fn test_consumer_penalty_degrades_to_blocking() {
const BLOCK_THRESHOLD: f64 = 0.15;
let engine = TrustEngine::new();
let peer = PeerId::random();
let failure_count = 10;
for _ in 0..failure_count {
engine.update_node_stats_weighted(&peer, NodeStatisticsUpdate::FailedResponse, 3.0);
}
let score = engine.score(&peer);
assert!(
score < BLOCK_THRESHOLD,
"after {failure_count} weight-3 failures, score {score} should be below block threshold {BLOCK_THRESHOLD}"
);
}
#[tokio::test]
async fn test_consumer_and_internal_events_combine() {
let engine = TrustEngine::new();
let peer = PeerId::random();
engine.update_node_stats(&peer, NodeStatisticsUpdate::CorrectResponse);
let after_success = engine.score(&peer);
assert!(
after_success > DEFAULT_NEUTRAL_TRUST,
"single success should raise above neutral"
);
engine.update_node_stats_weighted(&peer, NodeStatisticsUpdate::FailedResponse, 3.0);
let after_failure = engine.score(&peer);
assert!(
after_failure < after_success,
"weight-3 failure ({after_failure}) should outweigh weight-1 success ({after_success})"
);
assert!(
after_failure < DEFAULT_NEUTRAL_TRUST,
"net effect ({after_failure}) should be below neutral ({DEFAULT_NEUTRAL_TRUST})"
);
}
#[tokio::test]
async fn test_trust_query_reflects_all_event_sources() {
let engine = TrustEngine::new();
let peer = PeerId::random();
engine.update_node_stats(&peer, NodeStatisticsUpdate::CorrectResponse);
engine.update_node_stats_weighted(&peer, NodeStatisticsUpdate::CorrectResponse, 2.0);
engine.update_node_stats(&peer, NodeStatisticsUpdate::FailedResponse);
let score = engine.score(&peer);
assert!(
score > DEFAULT_NEUTRAL_TRUST,
"combined score {score} should be above neutral (net positive events)"
);
}
#[tokio::test]
async fn test_time_decay_applies_to_consumer_events() {
let engine = TrustEngine::new();
let peer = PeerId::random();
engine.update_node_stats_weighted(&peer, NodeStatisticsUpdate::FailedResponse, 3.0);
let after_failure = engine.score(&peer);
assert!(
after_failure < DEFAULT_NEUTRAL_TRUST,
"after failure, score {after_failure} should be below neutral"
);
let three_days = std::time::Duration::from_secs(3 * 24 * 3600);
engine.simulate_elapsed(&peer, three_days).await;
let after_decay = engine.score(&peer);
assert!(
after_decay > after_failure,
"score should decay toward neutral: {after_failure} -> {after_decay}"
);
let distance_from_neutral = (after_decay - DEFAULT_NEUTRAL_TRUST).abs();
assert!(
distance_from_neutral < 0.15,
"after 3 days, score {after_decay} should be near neutral (distance {distance_from_neutral})"
);
}
#[tokio::test]
async fn test_consumer_rewards_restore_trust_protection() {
const TRUST_PROTECTION_THRESHOLD: f64 = 0.7;
let engine = TrustEngine::new();
let peer = PeerId::random();
for _ in 0..5 {
engine.update_node_stats(&peer, NodeStatisticsUpdate::FailedResponse);
}
let low_score = engine.score(&peer);
assert!(
low_score < TRUST_PROTECTION_THRESHOLD,
"peer should start below trust protection: {low_score}"
);
let success_rounds = 30;
for _ in 0..success_rounds {
engine.update_node_stats_weighted(&peer, NodeStatisticsUpdate::CorrectResponse, 3.0);
}
let restored_score = engine.score(&peer);
assert!(
restored_score >= TRUST_PROTECTION_THRESHOLD,
"after {success_rounds} weight-3 successes, score {restored_score} should be >= {TRUST_PROTECTION_THRESHOLD}"
);
}
}