diff --git a/dash-spv/src/network/manager.rs b/dash-spv/src/network/manager.rs index 680eedf4d..6b5b40e2a 100644 --- a/dash-spv/src/network/manager.rs +++ b/dash-spv/src/network/manager.rs @@ -1,6 +1,6 @@ //! Peer network manager for SPV client -use std::collections::{HashMap, HashSet}; +use std::collections::HashMap; use std::net::{Ipv4Addr, SocketAddr, SocketAddrV4}; use std::path::PathBuf; use std::sync::atomic::{AtomicUsize, Ordering}; @@ -16,7 +16,7 @@ use crate::network::addrv2::AddrV2Handler; use crate::network::constants::*; use crate::network::discovery::DnsDiscovery; use crate::network::pool::PeerPool; -use crate::network::reputation::{ChangeReason, PeerReputationManager, ReputationAware}; +use crate::network::reputation::{ChangeReason, PeerReputationManager}; use crate::network::{ HandshakeManager, Message, MessageDispatcher, MessageType, NetworkEvent, NetworkManager, NetworkRequest, Peer, RequestSender, @@ -34,6 +34,31 @@ use tokio_util::sync::CancellationToken; const DEFAULT_NETWORK_EVENT_CAPACITY: usize = 10000; +/// Latency probes fired right after handshake to seed peer scoring. +const INITIAL_PROBE_PINGS: usize = 3; + +/// How long to wait for the initial probe pongs before scoring the peer. +const INITIAL_PROBE_WAIT: Duration = Duration::from_millis(1200); + +/// How often to probe known-but-unconnected peers looking for better ones. +const SCOUT_INTERVAL: Duration = Duration::from_secs(45); + +/// Peers to probe per scout tick. +const SCOUT_BATCH: usize = 4; + +/// Latency tier (lower is better): the band we rank and keep peers by. We only dial peers +/// in the best-known tier or one worse, and drop a connected peer more than one tier below +/// the best in the pool — so a genuinely slow peer never takes a slot. `<=50ms`=0, +/// `<=100ms`=1, `<=150ms`=2, else 3. +fn latency_tier(latency_ms: u32) -> u8 { + match latency_ms { + 0..=50 => 0, + 51..=100 => 1, + 101..=150 => 2, + _ => 3, + } +} + /// Peer network manager pub struct PeerNetworkManager { /// Peer pool @@ -68,16 +93,12 @@ pub struct PeerNetworkManager { capability_rejected: Arc>>, /// Cached count of currently connected peers for fast, non-blocking queries connected_peer_count: Arc, - /// Disable headers2 after decompression failure - headers2_disabled: Arc>>, /// Dispatcher for unbounded and message-type filtered message distribution. message_dispatcher: Arc>, /// Request queue sender, cloneable handle for sending requests to the network manager. request_tx: UnboundedSender, /// Request queue receiver (consumed by send loop). request_rx: Arc>>>, - /// Round-robin counter for distributing requests across peers. - round_robin_counter: Arc, /// Network event bus for notifying about network/peer related changes. network_event_sender: broadcast::Sender, } @@ -135,11 +156,9 @@ impl PeerNetworkManager { required_services, capability_rejected: Arc::new(RwLock::new(HashMap::new())), connected_peer_count: Arc::new(AtomicUsize::new(0)), - headers2_disabled: Arc::new(Mutex::new(HashSet::new())), message_dispatcher: Arc::new(Mutex::new(MessageDispatcher::default())), request_tx, request_rx: Arc::new(Mutex::new(Some(request_rx))), - round_robin_counter: Arc::new(AtomicUsize::new(0)), network_event_sender: broadcast::Sender::new(DEFAULT_NETWORK_EVENT_CAPACITY), }) } @@ -225,9 +244,9 @@ impl PeerNetworkManager { /// Connect to a specific peer async fn connect_to_peer(&self, addr: SocketAddr) { - // Check reputation first - if !self.reputation_manager.should_connect_to_peer(&addr).await { - tracing::warn!("Not connecting to {} due to bad reputation", addr); + // Skip peers whose multiplier has collapsed to zero (untrusted). + if !self.reputation_manager.is_usable(&addr).await { + tracing::warn!("Not connecting to {} - untrusted peer", addr); return; } @@ -241,9 +260,6 @@ impl PeerNetworkManager { return; // Already being connected to } - // Record connection attempt - self.reputation_manager.record_connection_attempt(addr).await; - let pool = self.pool.clone(); let network = self.network; let addrv2_handler = self.addrv2_handler.clone(); @@ -253,7 +269,6 @@ impl PeerNetworkManager { let required_services = self.required_services; let capability_rejected = self.capability_rejected.clone(); let connected_peer_count = self.connected_peer_count.clone(); - let headers2_disabled = self.headers2_disabled.clone(); let message_dispatcher = self.message_dispatcher.clone(); let network_event_sender = self.network_event_sender.clone(); @@ -305,14 +320,52 @@ impl PeerNetworkManager { } tracing::info!("Successfully connected to {}", addr); + // Seed behaviour and latency from persisted history, then measure + // latency inline so the peer enters the pool already scored and sync + // traffic avoids slow peers from the very first request. + if let Some((multiplier, median_ms)) = + reputation_manager.hint(&addr).await + { + peer.set_behavior_multiplier(multiplier); + if let Some(ms) = median_ms { + peer.record_rtt(Duration::from_millis(ms as u64)); + } + } + Self::probe_latency_inline(&mut peer).await; + + // Record the freshly-measured latency to reputation (so its tier + // is known at once) and reject a clearly-slower peer BEFORE it + // joins the pool, so a bad peer never receives any sync traffic. + // Good peers connect faster, so they are already measured here. + if let Some(rtt) = peer.median_rtt() { + let ms = rtt.as_millis() as u32; + reputation_manager.record_latency(addr, ms).await; + let connected = pool.get_connected_addresses().await; + let best = reputation_manager + .latencies(&connected) + .await + .values() + .map(|m| latency_tier(*m)) + .min(); + if let Some(best) = best { + if latency_tier(ms) > best + 1 { + tracing::info!( + "Rejecting peer {} (tier {} > best {} + 1) before pool — worse than the set", + addr, + latency_tier(ms), + best + ); + pool.remove_peer(&addr).await; + return; + } + } + } + // Request addresses from the peer for discovery if let Err(e) = peer.send_message(NetworkMessage::GetAddr).await { tracing::warn!("Failed to send GetAddr to {}: {}", addr, e); } - // Record successful connection - reputation_manager.record_successful_connection(addr).await; - // Add to pool if let Err(e) = pool.add_peer(addr, peer).await { tracing::error!("Failed to add peer to pool: {}", e); @@ -346,7 +399,6 @@ impl PeerNetworkManager { shutdown_token, reputation_manager.clone(), connected_peer_count.clone(), - headers2_disabled.clone(), message_dispatcher, network_event_sender.clone(), ) @@ -356,10 +408,7 @@ impl PeerNetworkManager { tracing::warn!("Handshake failed with {}: {}", addr, e); // Only clears connecting set. Peer was never added, so no count/event needed. pool.remove_peer(&addr).await; - // Update reputation for handshake failure - reputation_manager - .update_reputation(addr, ChangeReason::HandshakeFailed) - .await; + reputation_manager.penalize(addr, ChangeReason::HandshakeFailed).await; // For handshake failures, try again later tokio::time::sleep(RECONNECT_DELAY).await; } @@ -369,10 +418,7 @@ impl PeerNetworkManager { tracing::debug!("Failed to connect to {}: {}", addr, e); // Only clears connecting set. Peer was never added, so no count/event needed. pool.remove_peer(&addr).await; - // Minor reputation penalty for connection failure - reputation_manager - .update_reputation(addr, ChangeReason::ConnectionFailed) - .await; + reputation_manager.penalize(addr, ChangeReason::ConnectionFailed).await; } } }); @@ -426,7 +472,6 @@ impl PeerNetworkManager { shutdown_token: CancellationToken, reputation_manager: Arc, connected_peer_count: Arc, - headers2_disabled: Arc>>, message_dispatcher: Arc>, network_event_sender: broadcast::Sender, ) { @@ -611,6 +656,11 @@ impl PeerNetworkManager { headers.len(), addr ); + { + let mut g = peer.write().await; + g.note_response(); + g.apply_reason(ChangeReason::GoodResponse); + } // Forward as regular Headers message let headers_msg = NetworkMessage::Headers(headers); let message = Message::new(msg.peer_address(), headers_msg); @@ -618,20 +668,15 @@ impl PeerNetworkManager { continue; // Already sent, don't forward the original Headers2 } Err(e) => { + // A peer that opted into headers2 but sent data that fails + // to decompress served us invalid data - distrust it. tracing::error!( - "Headers2 decompression failed from {}: {} - disabling headers2", + "Headers2 decompression failed from {}: {} - distrusting peer", addr, e ); - headers2_disabled.lock().await.insert(addr); - // Apply reputation penalty - reputation_manager - .update_reputation( - addr, - ChangeReason::Headers2DecompressionFailed, - ) - .await; - continue; // Don't forward corrupted message + peer.write().await.apply_reason(ChangeReason::InvalidData); + break; // Disconnect the untrusted peer } } } @@ -670,6 +715,13 @@ impl PeerNetworkManager { } } + // A forwarded data message answers an outstanding request: + // clear the stall timer and reward the peer. + { + let mut peer_guard = peer.write().await; + peer_guard.note_response(); + peer_guard.apply_reason(ChangeReason::GoodResponse); + } message_dispatcher.lock().await.dispatch(&msg); } Ok(None) => { @@ -685,10 +737,7 @@ impl PeerNetworkManager { } NetworkError::Timeout => { tracing::debug!("Timeout reading from {}, continuing...", addr); - // Minor reputation penalty for timeout - reputation_manager - .update_reputation(addr, ChangeReason::ReadTimeout) - .await; + peer.write().await.apply_reason(ChangeReason::Timeout); continue; } _ => { @@ -703,13 +752,7 @@ impl PeerNetworkManager { "BLOCK DECODE FAILURE - Error details: {}", error_msg ); - // Reputation penalty for invalid data - reputation_manager - .update_reputation( - addr, - ChangeReason::InvalidTransactionInBlock, - ) - .await; + peer.write().await.apply_reason(ChangeReason::BadResponse); } else if error_msg .contains("Failed to decode transactions for block") { @@ -752,6 +795,19 @@ impl PeerNetworkManager { } } + // Persist the peer's final behaviour and latency before removing it. + if let Some(peer) = pool.get_peer(&addr).await { + let mut guard = peer.write().await; + // Reward a long, healthy connection. + if Duration::from_secs(60 * loop_iteration) > Duration::from_secs(3600) { + guard.apply_reason(ChangeReason::LongUptime); + } + let multiplier = guard.behavior_multiplier(); + let median = guard.median_rtt().map(|d| d.as_millis() as u32); + drop(guard); + reputation_manager.record(addr, multiplier, median).await; + } + // Remove from pool and notify consumers tracing::warn!("Disconnecting from {} (peer reader loop ended)", addr); Self::remove_peer_and_notify( @@ -761,15 +817,6 @@ impl PeerNetworkManager { &network_event_sender, ) .await; - - headers2_disabled.lock().await.remove(&addr); - - // Give small positive reputation if peer maintained long connection - let conn_duration = Duration::from_secs(60 * loop_iteration); // Rough estimate - if conn_duration > Duration::from_secs(3600) { - // 1 hour - reputation_manager.update_reputation(addr, ChangeReason::LongUptime).await; - } }); } @@ -798,25 +845,9 @@ impl PeerNetworkManager { match request { Some(NetworkRequest::SendMessage(msg)) => { tracing::debug!("Request processor: sending {}", msg.cmd()); - // Spawn each send concurrently to allow parallel requests across peers. let this = this.clone(); tokio::spawn(async move { - let result = match &msg { - // Distribute across peers for parallel sync - NetworkMessage::GetCFHeaders(_) - | NetworkMessage::GetCFilters(_) - | NetworkMessage::GetData(_) - | NetworkMessage::GetMnListD(_) - | NetworkMessage::GetQRInfo(_) - | NetworkMessage::GetHeaders(_) - | NetworkMessage::GetHeaders2(_) => { - this.send_distributed(msg).await - } - _ => { - this.send_to_single_peer(msg).await - } - }; - if let Err(e) = result { + if let Err(e) = this.pool.send(msg).await { tracing::error!("Request processor: failed to send message: {}", e); } }); @@ -826,24 +857,15 @@ impl PeerNetworkManager { let this = this.clone(); tokio::spawn(async move { let fallback_msg = msg.clone(); - let result = match this.pool.get_peer(&peer_address).await { - Some(peer) => match this.send_message_to_peer(&peer_address, &peer, msg).await { - Ok(()) => Ok(()), - Err(err) => { - tracing::warn!( - "Target peer {} send failed ({}), falling back to distributed send", - peer_address, - err - ); - this.send_distributed(fallback_msg).await - } - }, - None => { + let result = match this.pool.send_to(peer_address, msg).await { + Ok(()) => Ok(()), + Err(err) => { tracing::warn!( - "Target peer {} disconnected, falling back to distributed send", - peer_address + "Target peer {} send failed ({}), falling back to best peer", + peer_address, + err ); - this.send_distributed(fallback_msg).await + this.pool.send(fallback_msg).await.map(|_| ()) } }; if let Err(e) = result { @@ -947,11 +969,13 @@ impl PeerNetworkManager { // Keep the cached counter in sync with actual pool count self.connected_peer_count.store(count, Ordering::Relaxed); if self.exclusive_mode { - // In exclusive mode, only reconnect to originally specified peers - for addr in self.initial_peers.iter() { - if !self.pool.is_connected(addr).await && !self.pool.is_connecting(addr).await { + // Reconnect only to configured peers, and only those within the best tier band — + // a configured peer that measured clearly worse than the others is not redialed + // (max_peers is a ceiling). Unmeasured peers pass, so the first connect tries all. + for addr in self.within_tier_band(self.initial_peers.clone()).await { + if !self.pool.is_connected(&addr).await && !self.pool.is_connecting(&addr).await { tracing::info!("Reconnecting to exclusive peer: {}", addr); - self.connect_to_peer(*addr).await; + self.connect_to_peer(addr).await; } } } else { @@ -961,11 +985,12 @@ impl PeerNetworkManager { // Re-read count after potential churn so top-up sees the current pool size. let count = self.pool.peer_count().await; if count < self.max_peers { - // Try known addresses first, sorted by reputation - let known = self.addrv2_handler.get_known_addresses().await; let needed = self.max_peers.saturating_sub(count); - // Select best peers based on reputation - let best_peers = self.reputation_manager.select_best_peers(known, needed * 2).await; + // Dial known addresses best-first, restricted to the best tier band so we + // never top up with a clearly-slower peer than the ones we could have. + let known = self.known_socket_addrs().await; + let best_peers = + self.within_tier_band(self.reputation_manager.rank(known).await).await; let mut attempted = 0; for addr in best_peers { @@ -988,16 +1013,32 @@ impl PeerNetworkManager { return; } - // Send ping to all peers if needed and disconnect unresponsive ones + // Ping peers if needed, snapshot their measured quality for persistence, + // and disconnect unresponsive ones. + let mut low_quality: Vec = Vec::new(); for (addr, peer) in self.pool.get_all_peers().await { let mut peer_guard = peer.write().await; if peer_guard.should_ping() { if let Err(e) = peer_guard.send_ping().await { tracing::error!("Failed to ping {}: {}", addr, e); - // Update reputation for ping failure - self.reputation_manager.update_reputation(addr, ChangeReason::PingFailed).await; + peer_guard.apply_reason(ChangeReason::PingFailed); } } + // We ONLY replace a peer when it STALLS (goes silent on a request). We do not + // churn a merely-slow-but-responsive peer here: once syncing, every peer is + // under our own request load, so a high latency is expected. (Genuinely-bad + // peers are handled up front by measurement + `drop_worse_tier_peers`.) + let stalling = peer_guard.is_stalling(); + if stalling { + peer_guard.apply_reason(ChangeReason::Timeout); + low_quality.push(addr); + } + if peer_guard.median_rtt().is_some() { + let median = peer_guard.median_rtt().map(|d| d.as_millis() as u32); + self.reputation_manager + .record(addr, peer_guard.behavior_multiplier(), median) + .await; + } let has_expired = peer_guard.remove_expired_pings(); drop(peer_guard); if has_expired { @@ -1005,6 +1046,19 @@ impl PeerNetworkManager { } } + // Drop any connected peer whose measured tier is clearly worse than the best in the + // pool (the "bad peer" of a mixed set) — it should never hold a slot. Runs in every + // mode (the mixed-set case is a local/exclusive scenario). + self.drop_worse_tier_peers().await; + + // Replace peers that stalled, swapping in the best known unconnected peer and + // probing more to find a permanent one. + if !self.exclusive_mode { + for addr in low_quality { + self.replace_peer(addr).await; + } + } + // Only save known peers if not in exclusive mode if !self.exclusive_mode { let addresses = self.addrv2_handler.get_known_addresses().await; @@ -1050,6 +1104,229 @@ impl PeerNetworkManager { } } + /// Send a burst of probe pings and wait briefly for the pongs, recording the + /// RTTs. Unanswered probes are scored as the worst bucket so a silent or slow + /// peer enters the pool already deprioritised. + async fn probe_latency_inline(peer: &mut Peer) { + let mut pending = Vec::new(); + for _ in 0..INITIAL_PROBE_PINGS { + match peer.send_ping().await { + Ok(nonce) => pending.push(nonce), + Err(_) => break, + } + } + if pending.is_empty() { + return; + } + let deadline = Instant::now() + INITIAL_PROBE_WAIT; + let mut answered = 0; + while answered < pending.len() { + let remaining = deadline.saturating_duration_since(Instant::now()); + if remaining.is_zero() { + break; + } + match tokio::time::timeout(remaining, peer.receive_message()).await { + Ok(Ok(Some(message))) => match message.inner() { + NetworkMessage::Pong(n) if pending.contains(n) => { + let _ = peer.handle_pong(*n); + answered += 1; + } + NetworkMessage::Ping(p) => { + let _ = peer.handle_ping(*p).await; + } + _ => {} + }, + _ => break, + } + } + for _ in answered..pending.len() { + peer.record_rtt(INITIAL_PROBE_WAIT); + } + } + + /// Probe known-but-unconnected peers we have not measured yet, so the address + /// pool has fresh latency data ready to swap in as replacements. Cycles until + /// every known peer has been measured. + async fn scout_once(&self) { + if self.exclusive_mode { + return; + } + let candidates = self.unmeasured_candidates().await; + for addr in candidates.into_iter().take(SCOUT_BATCH) { + match self.probe_peer(addr).await { + Some(median_ms) => { + self.reputation_manager.record_latency(addr, median_ms).await; + tracing::debug!("Scout measured {} -> {}ms", addr, median_ms); + } + None => { + // Unreachable/failed probe: penalise so it ranks lower and, after + // enough failures, its multiplier hits zero and it drops out entirely. + self.reputation_manager.penalize(addr, ChangeReason::ConnectionFailed).await; + } + } + } + } + + /// Known addresses we are not connected to and have not measured yet. + async fn unmeasured_candidates(&self) -> Vec { + let mut out = Vec::new(); + for addr in self.known_socket_addrs().await { + if self.pool.is_connected(&addr).await + || self.pool.is_connecting(&addr).await + || !self.reputation_manager.is_usable(&addr).await + || self.reputation_manager.is_measured(&addr).await + || self.is_capability_rejected(&addr).await + { + continue; + } + out.push(addr); + } + out + } + + async fn known_socket_addrs(&self) -> Vec { + self.addrv2_handler + .get_known_addresses() + .await + .iter() + .filter_map(|m| m.socket_addr().ok()) + .collect() + } + + /// Keep only dial candidates within one latency tier of the best MEASURED candidate, + /// so we never add a clearly-slower peer while good ones are known (connect only to + /// the best tier or one worse). Unmeasured peers pass — they still need a probe. + async fn within_tier_band(&self, ranked: Vec) -> Vec { + let latencies = self.reputation_manager.latencies(&ranked).await; + let best_tier = + ranked.iter().filter_map(|a| latencies.get(a)).map(|ms| latency_tier(*ms)).min(); + let Some(best_tier) = best_tier else { + return ranked; // nothing measured yet — keep all to bootstrap + }; + ranked + .into_iter() + .filter(|a| { + latencies.get(a).map(|ms| latency_tier(*ms) <= best_tier + 1).unwrap_or(true) + }) + .collect() + } + + /// Drop connected peers whose latency tier is more than one worse than the best in the + /// pool. `max_peers` is a ceiling, not a quota: a genuinely-slow peer is not worth a + /// slot, so we run with fewer rather than keep it. Never drops below one peer, and if + /// every peer is equally bad nothing is dropped (nothing better to prefer). + async fn drop_worse_tier_peers(&self) { + let mut tiers: Vec<(SocketAddr, u8)> = Vec::new(); + for (addr, _) in self.pool.get_all_peers().await { + if let Some(ms) = self.reputation_manager.hint(&addr).await.and_then(|(_, l)| l) { + tiers.push((addr, latency_tier(ms))); + } + } + let Some(best) = tiers.iter().map(|(_, t)| *t).min() else { + return; + }; + for (addr, tier) in tiers { + if tier > best + 1 && self.pool.peer_count().await > 1 { + tracing::info!( + "Dropping peer {} (tier {} > best tier {} + 1) — worse than the pool", + addr, + tier, + best + ); + let _ = self.disconnect_peer(&addr, "worse latency tier than the pool").await; + } + } + } + + /// The best known peer we are not connected to (within the best tier band), for use + /// as a replacement. + async fn best_unconnected_peer(&self) -> Option { + let ranked = self + .within_tier_band(self.reputation_manager.rank(self.known_socket_addrs().await).await) + .await; + for addr in ranked { + if !self.pool.is_connected(&addr).await + && !self.pool.is_connecting(&addr).await + && !self.is_capability_rejected(&addr).await + { + return Some(addr); + } + } + None + } + + /// Drop a peer whose quality fell below the threshold, swap in the best known + /// unconnected peer as a temporary replacement, and probe more peers to find a + /// permanent one. Keeps the peer if it is our only one and nothing can replace it. + async fn replace_peer(&self, addr: SocketAddr) { + let replacement = self.best_unconnected_peer().await; + if replacement.is_none() && self.pool.peer_count().await <= 1 { + return; + } + let _ = self.disconnect_peer(&addr, "quality below threshold").await; + if let Some(candidate) = replacement { + tracing::info!("Replacing low-quality peer {} with {}", addr, candidate); + self.connect_to_peer(candidate).await; + } + let scout = self.clone(); + tokio::spawn(async move { scout.scout_once().await }); + } + + /// Connect, handshake and ping a peer once to measure its median latency, then + /// disconnect. Returns the median RTT in ms on success. + async fn probe_peer(&self, addr: SocketAddr) -> Option { + const PROBE_TOTAL: Duration = Duration::from_secs(8); + const PROBE_READ: Duration = Duration::from_millis(500); + + if !self.pool.mark_connecting(addr).await { + return None; + } + let result = self.probe_peer_inner(addr, PROBE_TOTAL, PROBE_READ).await; + // Clears the connecting flag; the probe peer was never added to the pool. + self.pool.remove_peer(&addr).await; + result + } + + async fn probe_peer_inner( + &self, + addr: SocketAddr, + total: Duration, + read_timeout: Duration, + ) -> Option { + let mut peer = tokio::select! { + r = Peer::connect(addr, CONNECTION_TIMEOUT.as_secs(), self.network) => r.ok()?, + _ = self.shutdown_token.cancelled() => return None, + }; + let mut handshake = HandshakeManager::new(self.network, self.user_agent.clone()); + tokio::select! { + r = handshake.perform_handshake(&mut peer) => r.ok()?, + _ = self.shutdown_token.cancelled() => return None, + }; + + let nonce = peer.send_ping().await.ok()?; + let start = Instant::now(); + while start.elapsed() < total { + match tokio::time::timeout(read_timeout, peer.receive_message()).await { + Ok(Ok(Some(message))) => match message.inner() { + NetworkMessage::Pong(n) if *n == nonce => { + let _ = peer.handle_pong(*n); + break; + } + NetworkMessage::Ping(p) => { + let _ = peer.handle_ping(*p).await; + } + _ => {} + }, + Ok(Ok(None)) | Err(_) => {} + Ok(Err(_)) => return None, + } + } + + let median = peer.median_rtt()?; + let _ = peer.disconnect().await; + Some(median.as_millis() as u32) + } + /// Start peer connection maintenance loop async fn start_maintenance_loop(&self) { let this = self.clone(); @@ -1060,6 +1337,9 @@ impl PeerNetworkManager { time::interval_at(Instant::now() + DNS_DISCOVERY_DELAY, DNS_DISCOVERY_DELAY); // Periodic reconnection check (active in both modes) let mut maintenance_interval = time::interval(MAINTENANCE_INTERVAL); + // Background probing of known-but-unconnected peers (non-exclusive mode). + let mut scout_interval = + time::interval_at(Instant::now() + SCOUT_INTERVAL, SCOUT_INTERVAL); let mut network_events = this.network_event_sender.subscribe(); while !this.shutdown_token.is_cancelled() { tokio::select! { @@ -1070,6 +1350,10 @@ impl PeerNetworkManager { _ = dns_interval.tick(), if !this.exclusive_mode => { this.dns_fallback_tick().await; } + _ = scout_interval.tick(), if !this.exclusive_mode => { + let scout = this.clone(); + tokio::spawn(async move { scout.scout_once().await }); + } event = network_events.recv() => { match event { Ok(event) => { @@ -1092,142 +1376,6 @@ impl PeerNetworkManager { }); } - /// Send a message to a single peer selected by message type requirements. - async fn send_to_single_peer(&self, message: NetworkMessage) -> NetworkResult<()> { - let peers = self.pool.get_all_peers().await; - - if peers.is_empty() { - return Err(NetworkError::ConnectionFailed("No connected peers".to_string())); - } - - let preferred_service = match &message { - NetworkMessage::FilterLoad(_) - | NetworkMessage::FilterClear - | NetworkMessage::MemPool => Some((ServiceFlags::BLOOM, true)), - NetworkMessage::GetCFHeaders(_) | NetworkMessage::GetCFilters(_) => { - Some((ServiceFlags::COMPACT_FILTERS, true)) - } - NetworkMessage::GetHeaders(_) | NetworkMessage::GetHeaders2(_) => { - Some((ServiceFlags::NODE_HEADERS_COMPRESSED, false)) - } - _ => None, - }; - - let (addr, peer) = if let Some((flags, required)) = preferred_service { - match self.pool.peer_with_service(flags).await { - Some((address, peer)) => { - tracing::debug!( - "Selected peer {} with {} for {}", - address, - flags, - message.cmd() - ); - (address, peer) - } - None if required => { - tracing::warn!("No peers support {}, cannot send {}", flags, message.cmd()); - return Err(NetworkError::ProtocolError(format!("No peers support {}", flags))); - } - None => self.next_peer(&peers), - } - } else { - self.next_peer(&peers) - }; - - self.send_message_to_peer(&addr, &peer, message).await - } - - /// Send a message distributed across connected peers using round-robin selection. - /// - /// Peer selection and message handling based on message type: - /// - Filters (GetCFHeaders/GetCFilters): requires peers that support compact filters - /// - Headers (GetHeaders/GetHeaders2): prefers headers2 peers, upgrades GetHeaders if supported - /// - Other (blocks, masternode data, etc.): uses all connected peers - async fn send_distributed(&self, message: NetworkMessage) -> NetworkResult<()> { - let peers = self.pool.get_all_peers().await; - - if peers.is_empty() { - return Err(NetworkError::ConnectionFailed("No connected peers".to_string())); - } - - // Select eligible peers based on message type - let (selected_peers, require_capability) = match &message { - NetworkMessage::GetCFHeaders(_) | NetworkMessage::GetCFilters(_) => { - let filter_peers = - self.pool.peers_with_service(ServiceFlags::COMPACT_FILTERS).await; - (filter_peers, true) - } - NetworkMessage::GetHeaders(_) | NetworkMessage::GetHeaders2(_) => { - // Prefer headers2 peers (excluding disabled), fall back to all - let disabled = self.headers2_disabled.lock().await; - let mut headers2_peers = - self.pool.peers_with_service(ServiceFlags::NODE_HEADERS_COMPRESSED).await; - headers2_peers.retain(|(addr, _)| !disabled.contains(addr)); - drop(disabled); - if headers2_peers.is_empty() { - (peers.clone(), false) - } else { - (headers2_peers, false) - } - } - _ => { - // All other messages use all connected peers - (peers.clone(), false) - } - }; - - if selected_peers.is_empty() { - return if require_capability { - Err(NetworkError::ProtocolError("No peers support required capability".to_string())) - } else { - Err(NetworkError::ConnectionFailed("No connected peers".to_string())) - }; - } - - let (addr, peer) = self.next_peer(&selected_peers); - - tracing::debug!("Distributing {} request to peer {}", message.cmd(), addr); - - self.send_message_to_peer(&addr, &peer, message).await - } - - /// Pick the next peer from `peers` using round-robin rotation. - fn next_peer( - &self, - peers: &[(SocketAddr, Arc>)], - ) -> (SocketAddr, Arc>) { - let idx = self.round_robin_counter.fetch_add(1, Ordering::Relaxed) % peers.len(); - (peers[idx].0, peers[idx].1.clone()) - } - - /// Send a message to the given peer. - /// For GetHeaders messages upgrade to GetHeaders2 if the peer supports it. - async fn send_message_to_peer( - &self, - addr: &SocketAddr, - peer: &Arc>, - message: NetworkMessage, - ) -> NetworkResult<()> { - let message = match message { - NetworkMessage::GetHeaders(get_headers) => { - let supports_headers2 = peer.read().await.can_request_headers2(); - if supports_headers2 && !self.headers2_disabled.lock().await.contains(addr) { - tracing::debug!("Upgrading GetHeaders to GetHeaders2 for peer {}", addr); - NetworkMessage::GetHeaders2(get_headers) - } else { - NetworkMessage::GetHeaders(get_headers) - } - } - other => other, - }; - - let mut peer_guard = peer.write().await; - peer_guard - .send_message(message) - .await - .map_err(|e| NetworkError::ProtocolError(format!("Failed to send to {}: {}", addr, e))) - } - /// Broadcast a message to all connected peers pub async fn broadcast(&self, message: NetworkMessage) -> Vec> { let peers = self.pool.get_all_peers().await; @@ -1282,30 +1430,6 @@ impl PeerNetworkManager { Ok(()) } - /// Get reputation information for all peers - pub async fn get_peer_reputations(&self) -> HashMap { - let reputations = self.reputation_manager.get_all_reputations().await; - reputations.into_iter().map(|(addr, rep)| (addr, (rep.score, rep.is_banned()))).collect() - } - - /// Ban a specific peer manually - pub async fn ban_peer(&self, addr: &SocketAddr, reason: &str) -> Result<(), Error> { - tracing::info!("Manually banning peer {} - reason: {}", addr, reason); - - // Disconnect the peer first - self.disconnect_peer(addr, reason).await?; - - // Update reputation to trigger ban - self.reputation_manager.update_reputation(*addr, ChangeReason::ManuallyBanned).await; - - Ok(()) - } - - /// Unban a specific peer - pub async fn unban_peer(&self, addr: &SocketAddr) { - self.reputation_manager.unban_peer(addr).await; - } - /// Shutdown the network manager pub async fn shutdown(&self) { tracing::info!("Shutting down peer network manager"); @@ -1392,11 +1516,9 @@ impl Clone for PeerNetworkManager { required_services: self.required_services, capability_rejected: self.capability_rejected.clone(), connected_peer_count: self.connected_peer_count.clone(), - headers2_disabled: self.headers2_disabled.clone(), message_dispatcher: self.message_dispatcher.clone(), request_tx: self.request_tx.clone(), request_rx: self.request_rx.clone(), - round_robin_counter: self.round_robin_counter.clone(), network_event_sender: self.network_event_sender.clone(), } } @@ -1423,19 +1545,17 @@ impl NetworkManager for PeerNetworkManager { } async fn send_message(&mut self, message: NetworkMessage) -> NetworkResult<()> { - // For sync messages that require consistent responses, send to only one peer + // For sync requests, route to the single best peer; broadcast everything else. match &message { NetworkMessage::GetHeaders(_) | NetworkMessage::GetHeaders2(_) | NetworkMessage::GetCFHeaders(_) | NetworkMessage::GetCFilters(_) | NetworkMessage::GetData(_) - | NetworkMessage::GetMnListD(_) => self.send_to_single_peer(message).await, + | NetworkMessage::GetMnListD(_) => self.pool.send(message).await.map(|_| ()), _ => { - // For other messages, broadcast to all peers let results = self.broadcast(message).await; - // Return error if all sends failed if results.is_empty() { return Err(NetworkError::ConnectionFailed("No connected peers".to_string())); } @@ -1513,11 +1633,9 @@ impl PeerNetworkManager { required_services, capability_rejected: Arc::new(RwLock::new(HashMap::new())), connected_peer_count: Arc::new(AtomicUsize::new(0)), - headers2_disabled: Arc::new(Mutex::new(HashSet::new())), message_dispatcher: Arc::new(Mutex::new(MessageDispatcher::default())), request_tx, request_rx: Arc::new(Mutex::new(Some(request_rx))), - round_robin_counter: Arc::new(AtomicUsize::new(0)), network_event_sender: broadcast::Sender::new(DEFAULT_NETWORK_EVENT_CAPACITY), } } diff --git a/dash-spv/src/network/peer.rs b/dash-spv/src/network/peer.rs index 2966ac52d..b97614141 100644 --- a/dash-spv/src/network/peer.rs +++ b/dash-spv/src/network/peer.rs @@ -1,10 +1,10 @@ //! Dash peer connection management. use dashcore::network::constants::ServiceFlags; -use std::collections::HashMap; +use std::collections::{HashMap, VecDeque}; use std::net::SocketAddr; use std::sync::Arc; -use std::time::{Duration, SystemTime}; +use std::time::{Duration, Instant, SystemTime}; use tokio::io::{AsyncReadExt, AsyncWriteExt}; use tokio::net::TcpStream; use tokio::sync::Mutex; @@ -15,6 +15,7 @@ use dashcore::Network; use crate::error::{NetworkError, NetworkResult}; use crate::network::constants::PING_INTERVAL; +use crate::network::reputation::ChangeReason; use crate::network::Message; /// Internal state for the TCP connection @@ -46,6 +47,27 @@ pub struct Peer { relay: Option, prefers_headers2: bool, sent_sendheaders2: bool, + headers2_disabled: bool, + rtt_samples: VecDeque, + behavior_multiplier: f32, + awaiting_since: Option, +} + +/// Number of recent response-time samples kept per peer for scoring. +const MAX_RTT_SAMPLES: usize = 8; + +/// A peer silent for this long after being sent a request is treated as stalling +/// and skipped, so retries go to a different peer — no second chances. +const STALL_THRESHOLD: Duration = Duration::from_secs(4); + +/// Bucket a response time into a 1..=4 quality score (4 = ideal, sub-50ms). +fn rtt_score(rtt: Duration) -> u8 { + match rtt.as_millis() { + 0..=49 => 4, + 50..=149 => 3, + 150..=399 => 2, + _ => 1, + } } impl Peer { @@ -72,6 +94,10 @@ impl Peer { relay: None, prefers_headers2: false, sent_sendheaders2: false, + headers2_disabled: false, + rtt_samples: VecDeque::new(), + behavior_multiplier: 1.0, + awaiting_since: None, } } @@ -118,6 +144,10 @@ impl Peer { relay: None, prefers_headers2: false, sent_sendheaders2: false, + headers2_disabled: false, + rtt_samples: VecDeque::new(), + behavior_multiplier: 1.0, + awaiting_since: None, }) } @@ -693,6 +723,7 @@ impl Peer { let rtt = now.duration_since(sent_time).unwrap_or(Duration::from_secs(0)); self.last_pong_received = Some(now); + self.record_rtt(rtt); tracing::debug!( "Received valid pong from {} with nonce {} (RTT: {:?})", @@ -790,11 +821,86 @@ impl Peer { self.sent_sendheaders2 } + /// Record a response-time sample used for peer quality scoring. + pub fn record_rtt(&mut self, rtt: Duration) { + self.rtt_samples.push_back(rtt); + while self.rtt_samples.len() > MAX_RTT_SAMPLES { + self.rtt_samples.pop_front(); + } + } + + /// Latency score in 1..=4 (higher is better): the worst bucket across recent + /// samples, so a peer that occasionally stalls is penalised. Unprobed peers + /// score 4 so they still get exercised until real samples arrive. + fn response_score(&self) -> u8 { + self.rtt_samples.iter().copied().map(rtt_score).min().unwrap_or(4) + } + + /// Overall selection quality in `[0, 4]`: the latency score scaled by the + /// behaviour multiplier, so a slow, misbehaving or lying peer ranks low. + pub fn quality(&self) -> f32 { + self.response_score() as f32 * self.behavior_multiplier + } + + /// Behaviour multiplier in `[0, 1]` (1 = trusted, 0 = never selected). + pub fn behavior_multiplier(&self) -> f32 { + self.behavior_multiplier + } + + /// Seed the behaviour multiplier from persisted history on connect. + pub fn set_behavior_multiplier(&mut self, multiplier: f32) { + self.behavior_multiplier = multiplier.clamp(0.0, 1.0); + } + + /// Fold a behaviour event into the multiplier (good raises, bad lowers, a + /// security violation zeroes it). + pub fn apply_reason(&mut self, reason: ChangeReason) { + self.behavior_multiplier = reason.apply(self.behavior_multiplier); + } + + /// Note that a request expecting a response was sent to this peer. + pub fn note_request_sent(&mut self) { + if self.awaiting_since.is_none() { + self.awaiting_since = Some(Instant::now()); + } + } + + /// Note that the peer answered with data, clearing any pending-request timer. + pub fn note_response(&mut self) { + self.awaiting_since = None; + } + + /// True when the peer was sent a request and has produced no data for longer than + /// `STALL_THRESHOLD` — it is stalling the sync and should not get more requests, and + /// the retry should go to a different peer. + pub fn is_stalling(&self) -> bool { + self.awaiting_since.map(|t| t.elapsed() > STALL_THRESHOLD).unwrap_or(false) + } + + /// Median recent response time, or `None` if the peer has not been probed. + pub fn median_rtt(&self) -> Option { + if self.rtt_samples.is_empty() { + return None; + } + let mut sorted: Vec = self.rtt_samples.iter().copied().collect(); + sorted.sort_unstable(); + Some(sorted[sorted.len() / 2]) + } + + /// Stop requesting compressed headers from this peer (e.g. after a + /// decompression failure); falls back to uncompressed `getheaders`. + pub fn disable_headers2(&mut self) { + self.headers2_disabled = true; + } + /// Check if we can request headers2 from this peer. pub fn can_request_headers2(&self) -> bool { // We can request headers2 if peer has the service flag for headers2 support // Note: We don't wait for SendHeaders2 from peer as that creates a race condition // during initial sync. The service flag is sufficient to know they support headers2. + if self.headers2_disabled { + return false; + } if let Some(services) = self.services { dashcore::network::constants::ServiceFlags::from(services) .has(dashcore::network::constants::NODE_HEADERS_COMPRESSED) @@ -847,4 +953,34 @@ mod tests { assert!(peer.remove_expired_pings()); assert!(peer.pending_pings.is_empty()); } + + #[test] + fn response_scoring() { + let addr: SocketAddr = "127.0.0.1:9999".parse().unwrap(); + let mut peer = Peer::dummy(addr); + + // Unprobed peers are optimistic and have no median. + assert_eq!(peer.response_score(), 4); + assert!(peer.median_rtt().is_none()); + + // A fast peer scores 4. + peer.record_rtt(Duration::from_millis(20)); + peer.record_rtt(Duration::from_millis(40)); + assert_eq!(peer.response_score(), 4); + assert_eq!(peer.median_rtt(), Some(Duration::from_millis(40))); + + // A single slow sample drags the score down to the worst bucket. + peer.record_rtt(Duration::from_millis(900)); + assert_eq!(peer.response_score(), 1); + } + + #[test] + fn rtt_samples_are_bounded() { + let addr: SocketAddr = "127.0.0.1:9999".parse().unwrap(); + let mut peer = Peer::dummy(addr); + for _ in 0..(super::MAX_RTT_SAMPLES + 5) { + peer.record_rtt(Duration::from_millis(10)); + } + assert_eq!(peer.rtt_samples.len(), super::MAX_RTT_SAMPLES); + } } diff --git a/dash-spv/src/network/pool.rs b/dash-spv/src/network/pool.rs index 1a2e68b21..11f2fc4d2 100644 --- a/dash-spv/src/network/pool.rs +++ b/dash-spv/src/network/pool.rs @@ -1,14 +1,47 @@ //! Peer pool for managing multiple peer connections -use crate::error::{NetworkError, SpvError as Error}; +use crate::error::{NetworkError, NetworkResult, SpvError as Error}; use crate::network::peer::Peer; use dashcore::network::constants::ServiceFlags; +use dashcore::network::message::NetworkMessage; use dashcore::prelude::CoreBlockHeight; use std::collections::{HashMap, HashSet}; use std::net::SocketAddr; use std::sync::Arc; +use std::time::Duration; use tokio::sync::RwLock; +/// Whether a message is a request that expects a response from the peer. +fn is_request(message: &NetworkMessage) -> bool { + matches!( + message, + NetworkMessage::GetHeaders(_) + | NetworkMessage::GetHeaders2(_) + | NetworkMessage::GetCFHeaders(_) + | NetworkMessage::GetCFilters(_) + | NetworkMessage::GetData(_) + | NetworkMessage::GetMnListD(_) + | NetworkMessage::GetQRInfo(_) + | NetworkMessage::MemPool + ) +} + +/// Service flags a message requires from a peer, and whether they are mandatory. +fn required_service(message: &NetworkMessage) -> Option<(ServiceFlags, bool)> { + match message { + NetworkMessage::FilterLoad(_) | NetworkMessage::FilterClear | NetworkMessage::MemPool => { + Some((ServiceFlags::BLOOM, true)) + } + NetworkMessage::GetCFHeaders(_) | NetworkMessage::GetCFilters(_) => { + Some((ServiceFlags::COMPACT_FILTERS, true)) + } + NetworkMessage::GetHeaders(_) | NetworkMessage::GetHeaders2(_) => { + Some((ServiceFlags::NODE_HEADERS_COMPRESSED, false)) + } + _ => None, + } +} + /// Pool for managing multiple peer instances pub struct PeerPool { /// Active peers mapped by address @@ -156,20 +189,6 @@ impl PeerPool { } } - /// Find the first connected peer that advertises the given service flags. - pub(crate) async fn peer_with_service( - &self, - flags: ServiceFlags, - ) -> Option<(SocketAddr, Arc>)> { - let peers = self.peers.read().await; - for (addr, peer) in peers.iter() { - if peer.read().await.has_service(flags) { - return Some((*addr, Arc::clone(peer))); - } - } - None - } - /// Collect all connected peers that advertise the given service flags. pub(crate) async fn peers_with_service( &self, @@ -196,6 +215,119 @@ impl PeerPool { false } + /// Send a message to the best-scoring peer able to serve it, returning the + /// address it was sent to. + /// + /// The pool owns peer selection: callers never hold a `Peer`. It picks the + /// probed peer with the best response-time bucket (breaking ties by lowest + /// median RTT), so a slow, stalling or still-unprobed peer is skipped. Sync + /// requests stay on a single fast peer, which the filter pipeline processes + /// faster than interleaved responses from several peers. + pub(crate) async fn send(&self, message: NetworkMessage) -> NetworkResult { + let candidates = match required_service(&message) { + Some((flags, must_have)) => { + let matching = self.peers_with_service(flags).await; + if matching.is_empty() { + if must_have { + return Err(NetworkError::ProtocolError(format!( + "No peers support {}", + flags + ))); + } + self.get_all_peers().await + } else { + matching + } + } + None => self.get_all_peers().await, + }; + + let (addr, peer) = self + .select_best(&candidates) + .await + .ok_or_else(|| NetworkError::ConnectionFailed("No connected peers".to_string()))?; + + Self::send_on_peer(&addr, &peer, message).await?; + Ok(addr) + } + + /// Send a message to a specific peer. Errors if the peer is gone so the + /// caller can fall back to `send`. + pub(crate) async fn send_to( + &self, + addr: SocketAddr, + message: NetworkMessage, + ) -> NetworkResult<()> { + let peer = self.get_peer(&addr).await.ok_or_else(|| { + NetworkError::ConnectionFailed(format!("Peer {} not connected", addr)) + })?; + Self::send_on_peer(&addr, &peer, message).await + } + + /// Pick the best peer: once any peer has a latency sample, only probed peers + /// are eligible (a still-unprobed or silent peer is skipped); among those it + /// takes the highest quality (latency score × behaviour multiplier), breaking + /// ties by lowest median RTT. + async fn select_best( + &self, + candidates: &[(SocketAddr, Arc>)], + ) -> Option<(SocketAddr, Arc>)> { + if candidates.is_empty() { + return None; + } + let mut entries = Vec::with_capacity(candidates.len()); + for (addr, peer) in candidates { + let (probed, stalling, quality, median) = { + let guard = peer.read().await; + ( + guard.median_rtt().is_some(), + guard.is_stalling(), + guard.quality(), + guard.median_rtt(), + ) + }; + entries.push(( + *addr, + peer.clone(), + probed, + stalling, + quality, + median.unwrap_or(Duration::MAX), + )); + } + + // Prefer non-stalling, probed peers; fall back only if none qualify so a + // single (even stalling) peer still gets used rather than stalling forever. + let any_probed = entries.iter().any(|e| e.2); + let any_live = entries.iter().any(|e| (!any_probed || e.2) && !e.3); + entries + .into_iter() + .filter(|e| (!any_probed || e.2) && (!any_live || !e.3)) + .max_by(|a, b| a.4.total_cmp(&b.4).then_with(|| b.5.cmp(&a.5))) + .map(|(addr, peer, ..)| (addr, peer)) + } + + async fn send_on_peer( + addr: &SocketAddr, + peer: &Arc>, + message: NetworkMessage, + ) -> NetworkResult<()> { + let mut guard = peer.write().await; + let message = match message { + NetworkMessage::GetHeaders(h) if guard.can_request_headers2() => { + NetworkMessage::GetHeaders2(h) + } + other => other, + }; + if is_request(&message) { + guard.note_request_sent(); + } + guard + .send_message(message) + .await + .map_err(|e| NetworkError::ProtocolError(format!("Failed to send to {}: {}", addr, e))) + } + /// Check if we need more peers pub async fn needs_more_peers(&self) -> bool { self.peer_count().await < self.max_peers @@ -247,6 +379,14 @@ impl PeerPool { peer.set_services(flags); self.peers.write().await.insert(addr, Arc::new(RwLock::new(peer))); } + + async fn insert_peer_with_rtts(&self, addr: SocketAddr, rtts: &[u64]) { + let mut peer = Peer::dummy(addr); + for &ms in rtts { + peer.record_rtt(std::time::Duration::from_millis(ms)); + } + self.peers.write().await.insert(addr, Arc::new(RwLock::new(peer))); + } } #[cfg(test)] @@ -269,6 +409,46 @@ mod tests { assert!(pool.is_connecting(&addr).await); } + #[tokio::test] + async fn test_select_skips_slow_peer() { + let pool = PeerPool::new(8); + let fast: SocketAddr = "127.0.0.1:2001".parse().unwrap(); + let slow: SocketAddr = "127.0.0.1:2002".parse().unwrap(); + pool.insert_peer_with_rtts(fast, &[20, 30, 40]).await; + pool.insert_peer_with_rtts(slow, &[900, 800]).await; + + let candidates = pool.get_all_peers().await; + let (addr, _) = pool.select_best(&candidates).await.unwrap(); + assert_eq!(addr, fast); + } + + #[tokio::test] + async fn test_select_breaks_ties_by_median() { + let pool = PeerPool::new(8); + let quick: SocketAddr = "127.0.0.1:2003".parse().unwrap(); + let quicker: SocketAddr = "127.0.0.1:2004".parse().unwrap(); + // Same worst-bucket score (4), so the lower median wins. + pool.insert_peer_with_rtts(quick, &[40, 45]).await; + pool.insert_peer_with_rtts(quicker, &[5, 10]).await; + + let candidates = pool.get_all_peers().await; + let (addr, _) = pool.select_best(&candidates).await.unwrap(); + assert_eq!(addr, quicker); + } + + #[tokio::test] + async fn test_select_prefers_probed_over_unprobed() { + let pool = PeerPool::new(8); + let probed: SocketAddr = "127.0.0.1:2005".parse().unwrap(); + let unprobed: SocketAddr = "127.0.0.1:2006".parse().unwrap(); + pool.insert_peer_with_rtts(probed, &[30]).await; + pool.insert_peer_with_rtts(unprobed, &[]).await; + + let candidates = pool.get_all_peers().await; + let (addr, _) = pool.select_best(&candidates).await.unwrap(); + assert_eq!(addr, probed); + } + #[tokio::test] async fn test_service_lookup() { let pool = PeerPool::new(8); @@ -276,13 +456,11 @@ mod tests { let combined = compact_filters | ServiceFlags::NODE_HEADERS_COMPRESSED; // No matches on empty pool - assert!(pool.peer_with_service(compact_filters).await.is_none()); assert!(pool.peers_with_service(compact_filters).await.is_empty()); // No matches when peers lack the requested flag let addr1: SocketAddr = "127.0.0.1:1001".parse().unwrap(); pool.insert_peer_with_services(addr1, ServiceFlags::NETWORK).await; - assert!(pool.peer_with_service(compact_filters).await.is_none()); assert!(pool.peers_with_service(compact_filters).await.is_empty()); // Single-flag lookup returns matching peers @@ -291,10 +469,6 @@ mod tests { pool.insert_peer_with_services(addr2, ServiceFlags::NETWORK | compact_filters).await; pool.insert_peer_with_services(addr3, ServiceFlags::NETWORK | combined).await; - let (found_addr, found_peer) = pool.peer_with_service(compact_filters).await.unwrap(); - assert!(found_addr == addr2 || found_addr == addr3); - assert!(found_peer.read().await.has_service(compact_filters)); - let filter_peers: HashMap = pool.peers_with_service(compact_filters).await.into_iter().collect(); assert_eq!(filter_peers.len(), 2); @@ -302,14 +476,11 @@ mod tests { assert!(filter_peers.contains_key(&addr3)); // Combined flags require all bits present - let (found_addr, _) = pool.peer_with_service(combined).await.unwrap(); - assert_eq!(found_addr, addr3); let combined_peers = pool.peers_with_service(combined).await; assert_eq!(combined_peers.len(), 1); assert_eq!(combined_peers[0].0, addr3); // NONE matches every peer in the pool - assert!(pool.peer_with_service(ServiceFlags::NONE).await.is_some()); let all = pool.peers_with_service(ServiceFlags::NONE).await; assert_eq!(all.len(), 3); } diff --git a/dash-spv/src/network/reputation.rs b/dash-spv/src/network/reputation.rs index f90656584..cdadf4960 100644 --- a/dash-spv/src/network/reputation.rs +++ b/dash-spv/src/network/reputation.rs @@ -1,225 +1,111 @@ -//! Peer reputation management system +//! Peer behaviour scoring for selection. //! -//! This module implements a reputation system to track peer behavior and protect -//! against malicious peers. It tracks both positive and negative behaviors, -//! implements automatic banning for excessive misbehavior, and provides reputation -//! decay over time for recovery. +//! Each peer carries a behaviour multiplier in `[0, 1]`. Good actions nudge it up +//! toward 1, bad actions down toward 0, and a security violation (the peer fed us +//! invalid data — it lied) drops it straight to 0. Peer selection multiplies this +//! by the peer's response-time score, so a misbehaving peer is deprioritised and a +//! lying one is never chosen. This manager persists the multiplier (and last +//! measured latency) so the judgement survives across connections. use crate::storage::PeerStorage; -use dashcore::network::address::AddrV2Message; use serde::{Deserialize, Deserializer, Serialize}; use std::collections::HashMap; use std::net::SocketAddr; use std::sync::Arc; -use std::time::{Duration, Instant}; use tokio::sync::RwLock; -/// Reason for a peer reputation change. Each reason owns its score delta -/// (positive = penalty, negative = reward) and a human-readable label. +/// A peer behaviour event. #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub enum ChangeReason { HandshakeFailed, ConnectionFailed, - Headers2DecompressionFailed, - ReadTimeout, + Timeout, PingFailed, - InvalidTransactionInBlock, - ManuallyBanned, + BadResponse, + GoodResponse, LongUptime, + /// The peer supplied invalid/untrustworthy data — treat it as a liar. + InvalidData, } impl ChangeReason { - /// Score delta for this reason: positive for misbehavior (penalty), - /// negative for good behavior (reward). - pub fn score(&self) -> i32 { + /// A security violation collapses the multiplier to zero immediately. + pub fn is_security_violation(&self) -> bool { + matches!(self, ChangeReason::InvalidData) + } + + /// Additive change to the `[0, 1]` behaviour multiplier for non-fatal events. + fn delta(&self) -> f32 { match self { - ChangeReason::HandshakeFailed => 10, - ChangeReason::ConnectionFailed => 2, - ChangeReason::Headers2DecompressionFailed => 10, - ChangeReason::ReadTimeout => 5, - ChangeReason::PingFailed => 5, - ChangeReason::InvalidTransactionInBlock => 20, - ChangeReason::ManuallyBanned => 100, - ChangeReason::LongUptime => -5, + ChangeReason::HandshakeFailed => -0.3, + ChangeReason::ConnectionFailed => -0.1, + ChangeReason::Timeout => -0.2, + ChangeReason::PingFailed => -0.2, + ChangeReason::BadResponse => -0.3, + ChangeReason::GoodResponse => 0.1, + ChangeReason::LongUptime => 0.1, + ChangeReason::InvalidData => -1.0, + } + } + + /// Apply this event to a multiplier, clamped to `[0, 1]` (0 on a security + /// violation). + pub fn apply(&self, multiplier: f32) -> f32 { + if self.is_security_violation() { + return 0.0; } + (multiplier + self.delta()).clamp(0.0, 1.0) } } impl std::fmt::Display for ChangeReason { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { let label = match self { - ChangeReason::HandshakeFailed => "Handshake failed", - ChangeReason::ConnectionFailed => "Connection failed", - ChangeReason::Headers2DecompressionFailed => "Headers2 decompression failed", - ChangeReason::ReadTimeout => "Read timeout", - ChangeReason::PingFailed => "Ping failed", - ChangeReason::InvalidTransactionInBlock => "Invalid transaction type in block", - ChangeReason::ManuallyBanned => "Manually banned", - ChangeReason::LongUptime => "Long connection uptime", + ChangeReason::HandshakeFailed => "handshake failed", + ChangeReason::ConnectionFailed => "connection failed", + ChangeReason::Timeout => "timed out", + ChangeReason::PingFailed => "ping failed", + ChangeReason::BadResponse => "bad response", + ChangeReason::GoodResponse => "good response", + ChangeReason::LongUptime => "long uptime", + ChangeReason::InvalidData => "invalid data", }; f.write_str(label) } } -/// Ban duration for misbehaving peers -const BAN_DURATION: Duration = Duration::from_secs(24 * 60 * 60); // 24 hours - -/// Reputation decay interval -const DECAY_INTERVAL: Duration = Duration::from_secs(60 * 60); // 1 hour - -/// Amount to decay reputation score per interval -const DECAY_AMOUNT: i32 = 5; - -/// Maximum misbehavior score before a peer is banned -const MAX_MISBEHAVIOR_SCORE: i32 = 100; - -/// Minimum score (most positive reputation) -const MIN_MISBEHAVIOR_SCORE: i32 = -50; - -const MAX_BAN_COUNT: u32 = 1000; - -const MAX_ACTION_COUNT: u64 = 1_000_000; - -fn clamp_peer_score<'de, D>(deserializer: D) -> Result -where - D: Deserializer<'de>, -{ - let mut v = i32::deserialize(deserializer)?; - - if v < MIN_MISBEHAVIOR_SCORE { - tracing::warn!("Peer has invalid score {v}, clamping to min {MIN_MISBEHAVIOR_SCORE}"); - v = MIN_MISBEHAVIOR_SCORE - } else if v > MAX_MISBEHAVIOR_SCORE { - tracing::warn!("Peer has invalid score {v}, clamping to max {MAX_MISBEHAVIOR_SCORE}"); - v = MAX_MISBEHAVIOR_SCORE - } - - Ok(v) +fn default_multiplier() -> f32 { + 1.0 } -fn clamp_peer_ban_count<'de, D>(deserializer: D) -> Result +fn clamp_multiplier<'de, D>(deserializer: D) -> Result where D: Deserializer<'de>, { - let mut v = u32::deserialize(deserializer)?; - - if v > MAX_BAN_COUNT { - tracing::warn!("Peer has excessive ban count {v}, clamping to {MAX_BAN_COUNT}"); - v = MAX_BAN_COUNT - } - - Ok(v) + Ok(f32::deserialize(deserializer)?.clamp(0.0, 1.0)) } -fn clamp_peer_connection_attempts<'de, D>(deserializer: D) -> Result -where - D: Deserializer<'de>, -{ - let mut v = u64::deserialize(deserializer)?; - - v = v.min(MAX_ACTION_COUNT); - - Ok(v) -} - -/// Peer reputation entry +/// Persisted peer quality: behaviour multiplier plus last measured latency. #[derive(Debug, Clone, Serialize, Deserialize)] pub struct PeerReputation { - /// Current misbehavior score - #[serde(deserialize_with = "clamp_peer_score")] - pub score: i32, - - /// Number of times this peer has been banned - #[serde(deserialize_with = "clamp_peer_ban_count")] - pub ban_count: u32, - - /// Time when the peer was banned (if currently banned) - #[serde(skip)] - pub banned_until: Option, - - /// Last time the reputation was updated - #[serde(skip, default = "Instant::now")] - pub last_update: Instant, - - /// Total number of positive actions - pub positive_actions: u64, - - /// Total number of negative actions - pub negative_actions: u64, - - /// Connection count - #[serde(deserialize_with = "clamp_peer_connection_attempts")] - pub connection_attempts: u64, - - /// Successful connection count - pub successful_connections: u64, - - /// Last connection time - #[serde(skip)] - pub last_connection: Option, + #[serde(default = "default_multiplier", deserialize_with = "clamp_multiplier")] + pub multiplier: f32, + #[serde(default)] + pub median_rtt_ms: Option, } impl Default for PeerReputation { fn default() -> Self { Self { - score: 0, - ban_count: 0, - banned_until: None, - last_update: Instant::now(), - positive_actions: 0, - negative_actions: 0, - connection_attempts: 0, - successful_connections: 0, - last_connection: None, + multiplier: 1.0, + median_rtt_ms: None, } } } -impl PeerReputation { - /// Check if the peer is currently banned - pub fn is_banned(&self) -> bool { - self.banned_until.is_some_and(|until| Instant::now() < until) - } - - /// Get remaining ban time - pub fn ban_time_remaining(&self) -> Option { - self.banned_until.and_then(|until| { - let now = Instant::now(); - if now < until { - Some(until - now) - } else { - None - } - }) - } - - /// Apply reputation decay - pub fn apply_decay(&mut self) { - let now = Instant::now(); - let elapsed = now - self.last_update; - - // Apply decay for each interval that has passed - let intervals = elapsed.as_secs() / DECAY_INTERVAL.as_secs(); - if intervals > 0 { - // Use saturating conversion to prevent overflow - // Cap at a reasonable maximum to avoid excessive decay - let intervals_i32 = intervals.min(i32::MAX as u64) as i32; - let decay = intervals_i32.saturating_mul(DECAY_AMOUNT); - self.score = (self.score - decay).max(MIN_MISBEHAVIOR_SCORE); - self.last_update = now; - } - - // Check if ban has expired - if self.is_banned() && self.ban_time_remaining().is_none() { - self.banned_until = None; - } - } -} - -/// Peer reputation manager +/// Tracks and persists per-peer behaviour multipliers. pub struct PeerReputationManager { - /// Reputation data for each peer - reputations: Arc>>, + records: Arc>>, } impl Default for PeerReputationManager { @@ -229,210 +115,96 @@ impl Default for PeerReputationManager { } impl PeerReputationManager { - /// Create a new reputation manager pub fn new() -> Self { Self { - reputations: Arc::new(RwLock::new(HashMap::new())), + records: Arc::new(RwLock::new(HashMap::new())), } } - /// Update peer reputation by the score delta of `reason`. - pub async fn update_reputation(&self, peer: SocketAddr, reason: ChangeReason) -> bool { - let score_change = reason.score(); - - let mut reputations = self.reputations.write().await; - let reputation = reputations.entry(peer).or_default(); - - // Apply decay first - reputation.apply_decay(); - - // Update score - let old_score = reputation.score; - reputation.score = - (reputation.score + score_change).clamp(MIN_MISBEHAVIOR_SCORE, MAX_MISBEHAVIOR_SCORE); - - // Track positive/negative actions - if score_change > 0 { - reputation.negative_actions += 1; - } else if score_change < 0 { - reputation.positive_actions += 1; - } - - // Check if peer should be banned - let should_ban = reputation.score >= MAX_MISBEHAVIOR_SCORE && !reputation.is_banned(); - if should_ban { - reputation.banned_until = Some(Instant::now() + BAN_DURATION); - reputation.ban_count += 1; - tracing::warn!( - "Peer {} banned for misbehavior (score: {}, ban #{}, reason: {})", - peer, - reputation.score, - reputation.ban_count, - reason - ); - } - - // Log significant changes - if score_change.abs() >= 10 || should_ban { - tracing::info!( - "Peer {} reputation changed: {} -> {} (change: {}, reason: {})", - peer, - old_score, - reputation.score, - score_change, - reason - ); - } - - should_ban + /// Apply a behaviour event to a peer that has no live connection to carry the + /// multiplier (e.g. a connect or handshake failure). Returns the new value. + pub async fn penalize(&self, addr: SocketAddr, reason: ChangeReason) -> f32 { + let mut records = self.records.write().await; + let record = records.entry(addr).or_default(); + record.multiplier = reason.apply(record.multiplier); + record.multiplier } - /// Check if a peer is banned - pub async fn is_banned(&self, peer: &SocketAddr) -> bool { - let mut reputations = self.reputations.write().await; - if let Some(reputation) = reputations.get_mut(peer) { - reputation.apply_decay(); - reputation.is_banned() - } else { - false + /// Write through a live peer's measured quality for persistence and reuse. + pub async fn record(&self, addr: SocketAddr, multiplier: f32, median_rtt_ms: Option) { + let mut records = self.records.write().await; + let record = records.entry(addr).or_default(); + record.multiplier = multiplier.clamp(0.0, 1.0); + if median_rtt_ms.is_some() { + record.median_rtt_ms = median_rtt_ms; } } - /// Record a connection attempt - pub async fn record_connection_attempt(&self, peer: SocketAddr) { - let mut reputations = self.reputations.write().await; - let reputation = reputations.entry(peer).or_default(); - reputation.connection_attempts += 1; - reputation.last_connection = Some(Instant::now()); + /// Record a peer's measured latency (from a background probe) without + /// touching its behaviour multiplier. + pub async fn record_latency(&self, addr: SocketAddr, median_rtt_ms: u32) { + let mut records = self.records.write().await; + records.entry(addr).or_default().median_rtt_ms = Some(median_rtt_ms); } - /// Record a successful connection - pub async fn record_successful_connection(&self, peer: SocketAddr) { - let mut reputations = self.reputations.write().await; - let reputation = reputations.entry(peer).or_default(); - reputation.successful_connections += 1; + /// Measured latency (ms) for each of `addrs` that has one, in a single lock. Used to + /// filter dial candidates and drop connected peers by latency tier. + pub async fn latencies(&self, addrs: &[SocketAddr]) -> HashMap { + let records = self.records.read().await; + addrs + .iter() + .filter_map(|a| records.get(a).and_then(|r| r.median_rtt_ms).map(|l| (*a, l))) + .collect() } - /// Get all peer reputations - pub async fn get_all_reputations(&self) -> HashMap { - let mut reputations = self.reputations.write().await; + /// Persisted `(multiplier, median_rtt_ms)` used to seed a peer on connect. + pub async fn hint(&self, addr: &SocketAddr) -> Option<(f32, Option)> { + self.records.read().await.get(addr).map(|r| (r.multiplier, r.median_rtt_ms)) + } - // Apply decay to all peers - for reputation in reputations.values_mut() { - reputation.apply_decay(); - } + /// A peer is dialable unless its multiplier has collapsed to zero. + pub async fn is_usable(&self, addr: &SocketAddr) -> bool { + self.records.read().await.get(addr).map(|r| r.multiplier > 0.0).unwrap_or(true) + } - reputations.clone() + /// True once the peer has a measured latency (has been probed at least once). + pub async fn is_measured(&self, addr: &SocketAddr) -> bool { + self.records.read().await.get(addr).map(|r| r.median_rtt_ms.is_some()).unwrap_or(false) } - /// Clear banned status for a peer (admin function) - pub async fn unban_peer(&self, peer: &SocketAddr) { - let mut reputations = self.reputations.write().await; - if let Some(reputation) = reputations.get_mut(peer) { - reputation.banned_until = None; - reputation.score = reputation.score.min(MAX_MISBEHAVIOR_SCORE - 10); - tracing::info!("Manually unbanned peer {}", peer); - } + /// Order candidate addresses best-first (highest multiplier, then lowest + /// measured latency), dropping any with a zero multiplier. + pub async fn rank(&self, mut addrs: Vec) -> Vec { + let records = self.records.read().await; + let key = |addr: &SocketAddr| -> (f32, u32) { + match records.get(addr) { + Some(r) => (r.multiplier, r.median_rtt_ms.unwrap_or(u32::MAX)), + None => (1.0, u32::MAX), + } + }; + addrs.retain(|addr| records.get(addr).map(|r| r.multiplier > 0.0).unwrap_or(true)); + addrs.sort_by(|a, b| { + let (ma, la) = key(a); + let (mb, lb) = key(b); + mb.total_cmp(&ma).then(la.cmp(&lb)) + }); + addrs } - /// Save reputation data to persistent storage pub async fn save_to_storage(&self, storage: &impl PeerStorage) -> std::io::Result<()> { - let reputations = self.reputations.read().await; - - storage.save_peers_reputation(&reputations).await.map_err(std::io::Error::other) + let records = self.records.read().await; + storage.save_peers_reputation(&records).await.map_err(std::io::Error::other) } - /// Load reputation data from persistent storage pub async fn load_from_storage(&self, storage: &impl PeerStorage) -> std::io::Result<()> { let data = storage.load_peers_reputation().await.map_err(std::io::Error::other)?; - - let mut reputations = self.reputations.write().await; - let mut loaded_count = 0; - let mut skipped_count = 0; - - for (addr, mut reputation) in data { - // Validate successful connections don't exceed attempts - reputation.successful_connections = - reputation.successful_connections.min(reputation.connection_attempts); - - // Skip entry if data appears corrupted - if reputation.positive_actions > MAX_ACTION_COUNT - || reputation.negative_actions > MAX_ACTION_COUNT - { - tracing::warn!("Skipping peer {} with potentially corrupted action counts", addr); - skipped_count += 1; - continue; - } - - // Apply initial decay based on ban count - if reputation.ban_count > 0 { - reputation.score = reputation.score.max(50); // Start with higher score for previously banned peers - } - - reputations.insert(addr, reputation); - loaded_count += 1; - } - - tracing::info!( - "Loaded reputation data for {} peers (skipped {} corrupted entries)", - loaded_count, - skipped_count - ); + let mut records = self.records.write().await; + *records = data; + tracing::info!("Loaded reputation data for {} peers", records.len()); Ok(()) } } -/// Helper trait for reputation-aware peer selection -pub trait ReputationAware { - /// Select best peers based on reputation - fn select_best_peers( - &self, - available_peers: Vec, - count: usize, - ) -> impl std::future::Future> + Send; - - /// Check if we should connect to a peer based on reputation - fn should_connect_to_peer( - &self, - peer: &SocketAddr, - ) -> impl std::future::Future + Send; -} - -impl ReputationAware for PeerReputationManager { - async fn select_best_peers( - &self, - available_peers: Vec, - count: usize, - ) -> Vec { - let mut peer_scores = Vec::new(); - let mut reputations = self.reputations.write().await; - - for peer in available_peers { - let Ok(socket_addr) = peer.socket_addr() else { - tracing::warn!("Skip invalid peer address: {:?}", peer); - continue; - }; - - let reputation = reputations.entry(socket_addr).or_default(); - reputation.apply_decay(); - - if !reputation.is_banned() { - peer_scores.push((socket_addr, reputation.score)); - } - } - - // Sort by score (lower is better) - peer_scores.sort_by_key(|(_, score)| *score); - - // Return the best peers - peer_scores.into_iter().take(count).map(|(peer, _)| peer).collect() - } - - async fn should_connect_to_peer(&self, peer: &SocketAddr) -> bool { - !self.is_banned(peer).await - } -} - // Include tests module #[cfg(test)] #[path = "reputation_tests.rs"] diff --git a/dash-spv/src/network/reputation_tests.rs b/dash-spv/src/network/reputation_tests.rs index 68b74e13b..ea3196587 100644 --- a/dash-spv/src/network/reputation_tests.rs +++ b/dash-spv/src/network/reputation_tests.rs @@ -1,109 +1,96 @@ -//! Unit tests for reputation system (in-module tests) +//! Unit tests for the peer behaviour multiplier. #[cfg(test)] mod tests { - use crate::storage::{PersistentPeerStorage, PersistentStorage}; - use super::super::*; + use crate::storage::{PersistentPeerStorage, PersistentStorage}; use std::net::SocketAddr; - async fn score(manager: &PeerReputationManager, peer: &SocketAddr) -> i32 { - manager.get_all_reputations().await.get(peer).map_or(0, |rep| rep.score) + fn addr(s: &str) -> SocketAddr { + s.parse().unwrap() } - #[tokio::test] - async fn test_basic_reputation_operations() { - let manager = PeerReputationManager::new(); - let peer: SocketAddr = "127.0.0.1:8333".parse().unwrap(); - - assert_eq!(score(&manager, &peer).await, 0); - - manager.update_reputation(peer, ChangeReason::HandshakeFailed).await; - assert_eq!(score(&manager, &peer).await, 10); + async fn multiplier(manager: &PeerReputationManager, peer: &SocketAddr) -> f32 { + manager.hint(peer).await.unwrap().0 + } - manager.update_reputation(peer, ChangeReason::LongUptime).await; - assert_eq!(score(&manager, &peer).await, 5); + #[test] + fn good_and_bad_actions_move_the_multiplier_within_bounds() { + // Starts trusted. + assert_eq!(ChangeReason::GoodResponse.apply(1.0), 1.0); + // A bad action lowers it. + let m = ChangeReason::Timeout.apply(1.0); + assert!(m < 1.0 && m > 0.0); + // Good actions raise it back, clamped at 1. + assert_eq!(ChangeReason::GoodResponse.apply(0.98), 1.0); + // It never goes below 0. + assert_eq!(ChangeReason::HandshakeFailed.apply(0.1), 0.0); } - #[tokio::test] - async fn test_banning_mechanism() { - let manager = PeerReputationManager::new(); - let peer: SocketAddr = "192.168.1.1:8333".parse().unwrap(); - - // Banned on the 10th violation (10 * 10 = 100). - for i in 0..10 { - let banned = manager.update_reputation(peer, ChangeReason::HandshakeFailed).await; - if i == 9 { - assert!(banned); - } else { - assert!(!banned); - } - } - - assert!(manager.is_banned(&peer).await); + #[test] + fn security_violation_zeroes_the_multiplier() { + assert_eq!(ChangeReason::InvalidData.apply(1.0), 0.0); + assert!(ChangeReason::InvalidData.is_security_violation()); } #[tokio::test] - async fn test_reputation_persistence() { + async fn penalize_accumulates_and_gates_usability() { let manager = PeerReputationManager::new(); - let peer1: SocketAddr = "10.0.0.1:8333".parse().unwrap(); - let peer2: SocketAddr = "10.0.0.2:8333".parse().unwrap(); - - manager.update_reputation(peer1, ChangeReason::LongUptime).await; - manager.update_reputation(peer1, ChangeReason::LongUptime).await; - manager.update_reputation(peer2, ChangeReason::InvalidTransactionInBlock).await; + let peer = addr("127.0.0.1:8333"); - let temp_dir = tempfile::TempDir::new().unwrap(); - let peer_storage = PersistentPeerStorage::open(temp_dir.path()) - .await - .expect("Failed to open PersistentPeerStorage"); - manager.save_to_storage(&peer_storage).await.unwrap(); + // Unknown peers are usable by default. + assert!(manager.is_usable(&peer).await); - let new_manager = PeerReputationManager::new(); - new_manager.load_from_storage(&peer_storage).await.unwrap(); + manager.penalize(peer, ChangeReason::Timeout).await; + assert!(multiplier(&manager, &peer).await < 1.0); + assert!(manager.is_usable(&peer).await); - assert_eq!(score(&new_manager, &peer1).await, -10); - assert_eq!(score(&new_manager, &peer2).await, 20); + manager.penalize(peer, ChangeReason::InvalidData).await; + assert_eq!(multiplier(&manager, &peer).await, 0.0); + assert!(!manager.is_usable(&peer).await); } #[tokio::test] - async fn test_peer_selection() { + async fn rank_orders_by_multiplier_then_latency_and_drops_zero() { let manager = PeerReputationManager::new(); + let best = addr("1.1.1.1:8333"); + let slower = addr("2.2.2.2:8333"); + let untrusted = addr("3.3.3.3:8333"); - let good_peer = AddrV2Message::dummy(0, "1.1.1.1".parse().unwrap(), 8333); - let neutral_peer = AddrV2Message::dummy(0, "2.2.2.2".parse().unwrap(), 8333); - let bad_peer = AddrV2Message::dummy(0, "3.3.3.3".parse().unwrap(), 8333); + manager.record(best, 1.0, Some(20)).await; + manager.record(slower, 1.0, Some(200)).await; + manager.record(untrusted, 0.0, Some(5)).await; - manager.update_reputation(good_peer.socket_addr().unwrap(), ChangeReason::LongUptime).await; - manager - .update_reputation( - bad_peer.socket_addr().unwrap(), - ChangeReason::InvalidTransactionInBlock, - ) - .await; + let ranked = manager.rank(vec![slower, untrusted, best]).await; + assert_eq!(ranked, vec![best, slower]); + } - let all_peers = vec![good_peer.clone(), neutral_peer.clone(), bad_peer.clone()]; - let selected = manager.select_best_peers(all_peers, 2).await; + #[tokio::test] + async fn record_latency_preserves_multiplier() { + let manager = PeerReputationManager::new(); + let peer = addr("10.0.0.1:8333"); + + manager.penalize(peer, ChangeReason::Timeout).await; + let m = multiplier(&manager, &peer).await; + manager.record_latency(peer, 42).await; - assert_eq!(selected.len(), 2); - assert_eq!(selected[0], good_peer.socket_addr().unwrap()); - assert_eq!(selected[1], neutral_peer.socket_addr().unwrap()); + assert_eq!(multiplier(&manager, &peer).await, m); + assert!(manager.is_measured(&peer).await); + assert_eq!(manager.hint(&peer).await, Some((m, Some(42)))); } #[tokio::test] - async fn test_connection_tracking() { + async fn persistence_round_trips_multiplier_and_latency() { let manager = PeerReputationManager::new(); - let peer: SocketAddr = "127.0.0.1:9999".parse().unwrap(); + let peer = addr("10.0.0.2:8333"); + manager.record(peer, 0.4, Some(120)).await; - // Track connection attempts - manager.record_connection_attempt(peer).await; - manager.record_connection_attempt(peer).await; - manager.record_successful_connection(peer).await; - - let reputations = manager.get_all_reputations().await; - let rep = &reputations[&peer]; + let temp_dir = tempfile::TempDir::new().unwrap(); + let storage = PersistentPeerStorage::open(temp_dir.path()).await.unwrap(); + manager.save_to_storage(&storage).await.unwrap(); - assert_eq!(rep.connection_attempts, 2); - assert_eq!(rep.successful_connections, 1); + let loaded = PeerReputationManager::new(); + loaded.load_from_storage(&storage).await.unwrap(); + assert_eq!(loaded.hint(&peer).await, Some((0.4, Some(120)))); } }