diff --git a/dash-spv/src/network/discovery.rs b/dash-spv/src/network/discovery.rs index 035bc7f28..74cb7bee7 100644 --- a/dash-spv/src/network/discovery.rs +++ b/dash-spv/src/network/discovery.rs @@ -12,6 +12,7 @@ //! Results from both sources are merged and deduplicated. use dashcore::Network; +use rand::seq::SliceRandom; use std::net::SocketAddr; /// DNS discovery for finding initial peers. @@ -60,6 +61,10 @@ impl DnsDiscovery { addresses.sort(); addresses.dedup(); + // Dedup needs the sort above, but a sorted list makes every client pick + // the same lowest-address peers via `take`/`truncate`, herding testers + // onto a handful of nodes. Shuffle so clients fan out across the set. + addresses.shuffle(&mut rand::thread_rng()); tracing::info!( "Discovered {} unique peer addresses for {:?} ({} from embedded seeds + DNS)", diff --git a/dash-spv/src/network/latency.rs b/dash-spv/src/network/latency.rs new file mode 100644 index 000000000..c9431f76e --- /dev/null +++ b/dash-spv/src/network/latency.rs @@ -0,0 +1,210 @@ +//! Per-peer response-latency tracking used to route sync requests. +//! +//! Routing quality is deliberately separate from the misbehavior score in +//! [`super::reputation`]. That score is an accusation: it only ever rises, and a +//! peer carrying no accusations is indistinguishable from one that has never been +//! tried. Routing needs the opposite property, evidence that expires back to +//! "unknown", so a peer that lost an early race is retried instead of starved. + +use std::collections::{HashMap, HashSet}; +use std::net::SocketAddr; +use std::time::{Duration, Instant}; + +/// Weight of the newest observation when folding it into a peer's mean, in +/// percent. Low enough that one slow response does not drop a peer out of +/// rotation, high enough that a peer going bad is demoted within a few requests. +const NEW_SAMPLE_WEIGHT: u32 = 30; + +/// How long an observation stays authoritative. Past this the peer counts as +/// unmeasured and is routed to again whatever its last mean, which is what stops +/// a peer from being frozen out permanently. This must exceed the stall timeout +/// plus a maintenance tick: a peer that is actually stalling then keeps refreshing +/// a bad mean and stays out of rotation, rather than aging back into it. +const SAMPLE_TTL: Duration = Duration::from_secs(30); + +/// A peer keeps receiving requests while its mean stays within this multiple of +/// the best measured peer's. +/// +/// Deliberately generous. Requests pipeline across peers, so a merely slower peer +/// still adds throughput and is worth keeping in rotation, and dropping it would +/// concentrate load on fewer peers for no gain. The job here is to shed peers that +/// have effectively stopped answering, which stand out by an order of magnitude +/// (a stall is recorded at ten seconds against sub-second healthy responses), not +/// to rank healthy peers against each other. +const SLOW_MULTIPLIER: u32 = 4; + +/// Absolute slack allowed alongside `SLOW_MULTIPLIER`, so that when peers are all +/// fast in absolute terms the multiple does not split hairs over normal jitter. +const SLOW_MARGIN: Duration = Duration::from_millis(500); + +#[derive(Debug, Clone, Copy)] +struct Observation { + mean: Duration, + at: Instant, +} + +/// Rolling response-time means for connected peers. +#[derive(Debug, Default)] +pub(super) struct PeerLatency { + observations: HashMap, +} + +impl PeerLatency { + /// Fold an observed response time into `addr`'s mean. + /// + /// A stalling request is recorded with the time it has been outstanding, which + /// is by definition at least the stall timeout, so a stalling peer's mean + /// climbs clear of any healthy peer's without needing a sentinel value. + pub(super) fn record(&mut self, addr: SocketAddr, elapsed: Duration) { + let now = Instant::now(); + match self.observations.get_mut(&addr) { + Some(observation) => { + observation.mean = blend(observation.mean, elapsed); + observation.at = now; + } + None => { + self.observations.insert( + addr, + Observation { + mean: elapsed, + at: now, + }, + ); + } + } + } + + /// Drop observations for peers that are no longer connected. + pub(super) fn retain_connected(&mut self, connected: &[SocketAddr]) { + self.observations.retain(|addr, _| connected.contains(addr)); + } + + /// The subset of `addrs` that should receive requests. + /// + /// Peers with no fresh observation are always eligible. We have no evidence + /// about them, and since only a request can produce evidence, excluding them + /// would make that lack of evidence permanent. The measured peers stay + /// eligible while they remain within reach of the best of them. + pub(super) fn eligible(&self, addrs: &[SocketAddr]) -> HashSet { + let now = Instant::now(); + let fresh = |addr: &SocketAddr| { + self.observations + .get(addr) + .filter(|observation| now.saturating_duration_since(observation.at) < SAMPLE_TTL) + }; + + let Some(best) = addrs.iter().filter_map(fresh).map(|observation| observation.mean).min() + else { + return addrs.iter().copied().collect(); + }; + let threshold = best.saturating_mul(SLOW_MULTIPLIER).max(best.saturating_add(SLOW_MARGIN)); + + addrs + .iter() + .copied() + .filter(|addr| fresh(addr).is_none_or(|observation| observation.mean <= threshold)) + .collect() + } +} + +/// Fold `sample` into `mean`, weighting the sample by `NEW_SAMPLE_WEIGHT`. +fn blend(mean: Duration, sample: Duration) -> Duration { + let old = mean.as_micros().saturating_mul(u128::from(100 - NEW_SAMPLE_WEIGHT)); + let new = sample.as_micros().saturating_mul(u128::from(NEW_SAMPLE_WEIGHT)); + let micros = (old.saturating_add(new) / 100).min(u128::from(u64::MAX)); + Duration::from_micros(micros as u64) +} + +#[cfg(test)] +mod tests { + use super::*; + + fn addr(last_octet: u8) -> SocketAddr { + SocketAddr::from(([127, 0, 0, last_octet], 9999)) + } + + #[test] + fn unmeasured_peers_are_always_eligible() { + let latency = PeerLatency::default(); + let peers = [addr(1), addr(2), addr(3)]; + + assert_eq!(latency.eligible(&peers).len(), 3); + } + + #[test] + fn slow_peer_drops_out_but_a_starved_peer_returns_when_its_sample_expires() { + let mut latency = PeerLatency::default(); + let (fast, slow) = (addr(1), addr(2)); + latency.record(fast, Duration::from_millis(50)); + latency.record(slow, Duration::from_secs(5)); + + let eligible = latency.eligible(&[fast, slow]); + assert!(eligible.contains(&fast)); + assert!(!eligible.contains(&slow), "a clearly slower peer stays out of rotation"); + + // Starvation is the failure this guards: once out of rotation the peer + // receives nothing, so only expiry can ever make it measurable again. + latency.observations.get_mut(&slow).expect("recorded above").at = + Instant::now() - SAMPLE_TTL - Duration::from_secs(1); + + assert!( + latency.eligible(&[fast, slow]).contains(&slow), + "an expired observation must put the peer back in rotation" + ); + } + + #[test] + fn comparable_peers_share_rotation() { + let mut latency = PeerLatency::default(); + let (a, b) = (addr(1), addr(2)); + latency.record(a, Duration::from_millis(20)); + latency.record(b, Duration::from_millis(90)); + + assert_eq!( + latency.eligible(&[a, b]).len(), + 2, + "peers that are all fast in absolute terms stay in rotation together" + ); + } + + #[test] + fn a_single_slow_response_does_not_evict_a_proven_peer() { + let mut latency = PeerLatency::default(); + let (fast, blip) = (addr(1), addr(2)); + latency.record(fast, Duration::from_millis(50)); + for _ in 0..10 { + latency.record(blip, Duration::from_millis(50)); + } + latency.record(blip, Duration::from_secs(1)); + + assert!( + latency.eligible(&[fast, blip]).contains(&blip), + "one blip should not undo a peer's track record" + ); + } + + #[test] + fn repeated_stalls_demote_a_peer() { + let mut latency = PeerLatency::default(); + let (fast, stalling) = (addr(1), addr(2)); + latency.record(fast, Duration::from_millis(50)); + latency.record(stalling, Duration::from_millis(50)); + for _ in 0..3 { + latency.record(stalling, Duration::from_secs(10)); + } + + assert!(!latency.eligible(&[fast, stalling]).contains(&stalling)); + } + + #[test] + fn disconnected_peers_are_forgotten() { + let mut latency = PeerLatency::default(); + latency.record(addr(1), Duration::from_millis(50)); + latency.record(addr(2), Duration::from_millis(50)); + + latency.retain_connected(&[addr(1)]); + + assert_eq!(latency.observations.len(), 1); + assert!(latency.observations.contains_key(&addr(1))); + } +} diff --git a/dash-spv/src/network/manager.rs b/dash-spv/src/network/manager.rs index dbfccb443..b4c2ec54b 100644 --- a/dash-spv/src/network/manager.rs +++ b/dash-spv/src/network/manager.rs @@ -15,6 +15,7 @@ use crate::error::{NetworkError, NetworkResult, SpvError as Error}; use crate::network::addrv2::AddrV2Handler; use crate::network::constants::*; use crate::network::discovery::DnsDiscovery; +use crate::network::latency::PeerLatency; use crate::network::pool::PeerPool; use crate::network::reputation::{ChangeReason, PeerReputationManager, ReputationAware}; use crate::network::{ @@ -26,6 +27,7 @@ use async_trait::async_trait; use dashcore::network::address::{AddrV2, AddrV2Message}; use dashcore::network::constants::ServiceFlags; use dashcore::network::message::NetworkMessage; +use dashcore::network::message_blockdata::Inventory; use dashcore::network::message_headers2::CompressionState; use dashcore::Network; use tokio::sync::mpsc::{unbounded_channel, UnboundedReceiver, UnboundedSender}; @@ -80,10 +82,69 @@ pub struct PeerNetworkManager { round_robin_counter: Arc, /// Network event bus for notifying about network/peer related changes. network_event_sender: broadcast::Sender, + /// Instant each peer's oldest unanswered request of a given kind was sent. + /// The answering response clears the entry, a stale one marks a stalling peer. + outstanding_requests: Arc>>, + /// Rolling response times per peer, used to route requests to the peers that + /// are actually answering fastest. + latency: Arc>, + /// Instant of the previous stall sweep, used to detect a suspend/resume gap so + /// a burst of stale requests on resume does not penalize every peer at once. + last_maintenance_at: Arc>, + /// Instant of the last reputation-driven eviction, enforcing a cooldown so a + /// replacement can prove itself before another peer is dropped. + last_eviction_at: Arc>>, } const CAPABILITY_REJECTED_TTL: Duration = Duration::from_secs(30 * 60); +/// A connected peer whose oldest sync request stays unanswered this long is +/// treated as stalling and penalized. Any response resets its timer, so a peer +/// streaming a large block is not punished for the download taking a while. +/// +/// Only kinds whose response time is a fair measure of the peer are judged here +/// (see `RequestKind::response_time_is_fair`), so this is generous: a filter, +/// filter-header or header response is at most a few tens of KB and arrives in +/// well under a second on any usable link. Large payloads are excluded, since +/// their transfer can legitimately exceed this on a slow mobile link. +pub(super) const REQUEST_STALL_TIMEOUT: Duration = Duration::from_secs(10); + +/// A peer whose request of some kind has gone unanswered this long stops being +/// picked for further requests of that same kind, until it answers. +/// +/// This is routing only, never a penalty, so it can be far stricter than +/// `REQUEST_STALL_TIMEOUT` without risking an honest peer: the cost of skipping +/// one wrongly is that another peer serves the request instead. It must stay +/// below the sync layer's own retry timeouts, so that a peer which dropped a +/// request is already out of the rotation by the time that request is reissued. +pub(super) const REQUEST_OWED_TIMEOUT: Duration = Duration::from_secs(5); + +/// If a maintenance sweep runs at least this long after the previous one, the +/// process was likely suspended (e.g. an iOS app backgrounded). Every +/// outstanding request would look stale at once, so the sweep refreshes their +/// timers and skips penalties for that round instead of blaming every peer. +const SUSPEND_GAP: Duration = Duration::from_secs(90); + +/// A connected peer scoring at least this becomes an eviction candidate: two +/// consecutive stalls, or equivalent misbehavior. Well below the +100 ban line, +/// so a peer is dropped and replaced long before it would be banned outright, and +/// eviction stays a soft demotion (the peer keeps its address-book entry and +/// decays back). +/// +/// Reachable only because a stalling peer's timer is re-armed rather than +/// dropped: one stall already records a mean response time that freezes the peer +/// out of routing, so it receives no new request to re-arm from and could +/// otherwise never earn a second strike. +const STUCK_PEER_EVICTION_SCORE: i32 = 20; + +/// Never evict a peer for reputation while at or below this many connections, so +/// a small pool is never churned down toward zero usable peers. +const MIN_CONNECTED_FLOOR: usize = 2; + +/// Minimum spacing between reputation-driven evictions, so a fresh replacement +/// has time to connect and prove itself before another peer is dropped. +const EVICTION_COOLDOWN: Duration = Duration::from_secs(30); + fn required_services_from_config(config: &ClientConfig, exclusive_mode: bool) -> ServiceFlags { if exclusive_mode { return ServiceFlags::NONE; @@ -141,6 +202,10 @@ impl PeerNetworkManager { 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), + outstanding_requests: Arc::new(Mutex::new(HashMap::new())), + latency: Arc::new(Mutex::new(PeerLatency::default())), + last_maintenance_at: Arc::new(Mutex::new(Instant::now())), + last_eviction_at: Arc::new(Mutex::new(None)), }) } @@ -249,6 +314,8 @@ impl PeerNetworkManager { let addrv2_handler = self.addrv2_handler.clone(); let shutdown_token = self.shutdown_token.clone(); let reputation_manager = self.reputation_manager.clone(); + let outstanding_requests = self.outstanding_requests.clone(); + let latency = self.latency.clone(); let user_agent = self.user_agent.clone(); let required_services = self.required_services; let capability_rejected = self.capability_rejected.clone(); @@ -313,9 +380,14 @@ impl PeerNetworkManager { // Record successful connection reputation_manager.record_successful_connection(addr).await; - // Add to pool + // Both ways this can fail, a full pool and an address + // already connected, are ordinary outcomes of dialling + // several candidates for the same slot: one wins and the + // rest arrive to find it taken. Topping up after an + // eviction does exactly that, so logging it as an error + // reports routine contention as a fault. if let Err(e) = pool.add_peer(addr, peer).await { - tracing::error!("Failed to add peer to pool: {}", e); + tracing::debug!("Not adding peer {} to pool: {}", addr, e); return; } @@ -345,6 +417,8 @@ impl PeerNetworkManager { addrv2_handler, shutdown_token, reputation_manager.clone(), + outstanding_requests, + latency, connected_peer_count.clone(), headers2_disabled.clone(), message_dispatcher, @@ -425,6 +499,8 @@ impl PeerNetworkManager { addrv2_handler: Arc, shutdown_token: CancellationToken, reputation_manager: Arc, + outstanding_requests: Arc>>, + latency: Arc>, connected_peer_count: Arc, headers2_disabled: Arc>>, message_dispatcher: Arc>, @@ -485,6 +561,22 @@ impl PeerNetworkManager { // Log all received messages at debug level to help troubleshoot tracing::trace!("Received {:?} from {}", msg.cmd(), addr); + // A substantive response clearing a tracked request is the one + // point where a peer's speed is observable, so time it here. + // The elapsed time runs from the oldest request still + // unanswered, so it counts queueing behind that peer's own + // backlog as well as its service time. That is deliberate: + // both delay us equally, and it lets a peer we are + // over-feeding report itself as congested. + if let Some(kind) = timed_response_kind(msg.inner()) { + let sent = outstanding_requests.lock().await.remove(&(addr, kind)); + if let Some(sent) = sent { + if kind.response_time_is_fair() { + latency.lock().await.record(addr, sent.elapsed()); + } + } + } + // Handle some messages directly match &msg.inner() { NetworkMessage::SendAddrV2 => { @@ -684,11 +776,9 @@ impl PeerNetworkManager { break; } NetworkError::Timeout => { + // Idle socket reads time out constantly on healthy + // connections, so this is not a quality signal. tracing::debug!("Timeout reading from {}, continuing...", addr); - // Minor reputation penalty for timeout - reputation_manager - .update_reputation(addr, ChangeReason::ReadTimeout) - .await; continue; } _ => { @@ -763,13 +853,6 @@ impl PeerNetworkManager { .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; - } }); } @@ -927,6 +1010,159 @@ impl PeerNetworkManager { } } + /// Penalize connected peers that have left a sync request unanswered past the + /// stall timeout, so they fall out of routing and become eviction candidates. + /// A long gap since the previous sweep is treated as a process suspend and the + /// timers are refreshed instead, so a backgrounded client does not blame every + /// peer at once on resume. + /// Returns true if this tick was treated as a suspend/resume grace tick, so + /// the caller can skip eviction rather than acting on stale pre-suspend state. + async fn sweep_stalled_peers(&self) -> bool { + let now = Instant::now(); + let gap = { + let mut last = self.last_maintenance_at.lock().await; + let gap = now.saturating_duration_since(*last); + *last = now; + gap + }; + + // Snapshot connected peers before taking the outstanding lock so this path + // never holds the outstanding lock while touching the pool lock. + let connected = self.pool.get_connected_addresses().await; + self.latency.lock().await.retain_connected(&connected); + + let stalled = { + let mut outstanding = self.outstanding_requests.lock().await; + outstanding.retain(|(addr, _), _| connected.contains(addr)); + + if gap > SUSPEND_GAP { + for sent in outstanding.values_mut() { + *sent = now; + } + return true; + } + + let aged: Vec<((SocketAddr, RequestKind), Duration)> = outstanding + .iter() + .filter(|((_, kind), _)| kind.response_time_is_fair()) + .filter_map(|(key, sent)| { + let waited = now.saturating_duration_since(*sent); + (waited > REQUEST_STALL_TIMEOUT).then_some((*key, waited)) + }) + .collect(); + // Re-arm rather than drop: one stall already freezes the peer out of + // routing, so it gets no new request to re-arm the timer, and a still + // unanswered request must keep counting for it to ever be evicted. + for (key, _) in &aged { + outstanding.insert(*key, now); + } + + // Collapse to one entry per peer, keeping its worst wait. A peer + // stalling on several kinds at once is one failing peer, not several, + // and must not take several strikes from a single sweep. + let mut stalled: HashMap = HashMap::new(); + for ((addr, _), waited) in aged { + let worst = stalled.entry(addr).or_insert(waited); + *worst = (*worst).max(waited); + } + stalled + }; + + // Recording how long the request has gone unanswered keeps the stalling + // peer's measured mean fresh as well as bad, so it stays out of rotation + // instead of ageing back into it while it is still failing to answer. + { + let mut latency = self.latency.lock().await; + for (addr, waited) in &stalled { + latency.record(*addr, *waited); + } + } + + // A peer already at the eviction threshold is condemned, so further strikes + // add nothing and would only march it toward an outright ban. + let scores = self.reputation_manager.scores_for(stalled.keys().copied()).await; + for (addr, _) in stalled { + if scores.get(&addr).copied().unwrap_or(0) >= STUCK_PEER_EVICTION_SCORE { + continue; + } + tracing::debug!("Peer {} stalled on a sync request, penalizing", addr); + self.reputation_manager.update_reputation(addr, ChangeReason::RequestTimeout).await; + } + false + } + + /// Evict the single worst-scoring connected peer when it is clearly bad and a + /// healthy replacement is available, so the client stops staying stuck on a + /// peer that stalls. Heavily guarded so a small pool is never churned toward + /// zero: it keeps a connection floor, only acts on a full pool with a + /// replacement ready, skips when every peer is equally bad (a network or + /// device problem, not a peer problem), never drops the sole peer providing a + /// required service, and evicts at most one peer per cooldown. + async fn evict_worst_stuck_peer(&self) { + if let Some(t) = *self.last_eviction_at.lock().await { + if Instant::now().saturating_duration_since(t) < EVICTION_COOLDOWN { + return; + } + } + + let connected = self.pool.get_connected_addresses().await; + let floor = self.max_peers.min(MIN_CONNECTED_FLOOR); + if connected.len() < self.max_peers || connected.len() <= floor { + return; + } + + // A replacement must be ready, or eviction is pure loss. + let mut has_replacement = false; + for known in self.addrv2_handler.get_known_addresses().await { + let Ok(sa) = known.socket_addr() else { + continue; + }; + if !connected.contains(&sa) && !self.is_capability_rejected(&sa).await { + has_replacement = true; + break; + } + } + if !has_replacement { + return; + } + + let scores = self.reputation_manager.scores_for(connected.iter().copied()).await; + let Some(min_score) = scores.values().copied().min() else { + return; + }; + let Some((worst_addr, worst_score)) = scores.into_iter().max_by_key(|(_, s)| *s) else { + return; + }; + + // The worst must be clearly bad, and at least one peer clearly better, + // otherwise the whole pool is bad and the problem is not this peer. + if worst_score < STUCK_PEER_EVICTION_SCORE || min_score >= STUCK_PEER_EVICTION_SCORE { + return; + } + + if self.is_sole_service_provider(&worst_addr).await { + return; + } + + tracing::info!( + "Evicting stalled peer {} (score {}) so a fresh peer can replace it", + worst_addr, + worst_score + ); + let _ = self.disconnect_peer(&worst_addr, "poor reputation (stalled requests)").await; + *self.last_eviction_at.lock().await = Some(Instant::now()); + } + + /// Whether `addr` is the only connected peer advertising the required service, + /// so evicting it would strand a sync phase with no capable peer. + async fn is_sole_service_provider(&self, addr: &SocketAddr) -> bool { + if self.required_services == ServiceFlags::NONE { + return false; + } + let providers = self.pool.peers_with_service(self.required_services).await; + providers.len() == 1 && providers[0].0 == *addr + } + async fn maintenance_tick(&self) { // Remove peers that the reader loop failed to clean up. // This should not trigger under normal operation. @@ -955,9 +1191,17 @@ impl PeerNetworkManager { } } } else { + // Penalize peers stalling on sync requests so they drop out of routing + // and become eviction candidates before the top-up below. + let grace_tick = self.sweep_stalled_peers().await; // Evict peers that lack required services before top-up so replacements // can be pulled in during the same tick. self.evict_mismatched_peers().await; + // Drop one clearly-bad stalling peer, but never on a resume grace tick + // where scores reflect stale pre-suspend state. + if !grace_tick { + self.evict_worst_stuck_peer().await; + } // 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 { @@ -1128,10 +1372,10 @@ impl PeerNetworkManager { tracing::warn!("No peers support {}, cannot send {}", flags, message.cmd()); return Err(NetworkError::ProtocolError(format!("No peers support {}", flags))); } - None => self.next_peer(&peers), + None => self.next_peer(&peers).await, } } else { - self.next_peer(&peers) + self.next_peer(&peers).await }; self.send_message_to_peer(&addr, &peer, message).await @@ -1184,20 +1428,92 @@ impl PeerNetworkManager { }; } - let (addr, peer) = self.next_peer(&selected_peers); + let selected_peers = match tracked_request_kind(&message) { + Some(kind) => self.without_peers_owing(kind, selected_peers).await, + None => selected_peers, + }; + + let (addr, peer) = self.next_peer(&selected_peers).await; tracing::trace!("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( + /// Drop peers that already owe us a response of `kind` from the candidates. + /// + /// A peer that silently discards a request is invisible to latency-based + /// routing: latency only ever records a response, so a peer that answers + /// nothing is never measured and keeps its turn in the rotation. It then wins + /// the retry of the very request it just dropped, which is how one such peer + /// turned a 30s block timeout into 60s and 120s stalls, with filter sync + /// halted behind it the whole time. + /// + /// The bar is what the peer owes rather than what it did wrong, so this needs + /// no judgement about how slow is too slow and never accuses an honest peer on + /// a slow link. It is also self-clearing: the peer becomes a candidate again + /// the moment it answers, or the moment the request is satisfied elsewhere. + /// + /// Skipping is only ever a preference. If every candidate owes us something, + /// the full set is kept, because refusing to send is worse than sending to a + /// busy peer. + async fn without_peers_owing( + &self, + kind: RequestKind, + peers: Vec<(SocketAddr, Arc>)>, + ) -> Vec<(SocketAddr, Arc>)> { + let now = Instant::now(); + let owing: HashSet = { + let outstanding = self.outstanding_requests.lock().await; + outstanding + .iter() + .filter(|((_, k), sent)| { + *k == kind && now.saturating_duration_since(**sent) > REQUEST_OWED_TIMEOUT + }) + .map(|((addr, _), _)| *addr) + .collect() + }; + + if owing.is_empty() { + return peers; + } + let free: Vec<(SocketAddr, Arc>)> = + peers.iter().filter(|(addr, _)| !owing.contains(addr)).cloned().collect(); + if free.is_empty() { + return peers; + } + free + } + + /// Pick a peer from `peers`, rotating over those answering fastest. + /// + /// Selection reads measured response times rather than the misbehavior score. + /// A score only ever accuses a peer, so a peer that is merely untried scores + /// the same as a good one, and rewarding responses to break that tie makes the + /// first peer to answer pull away and take the whole rotation: it earns more + /// traffic, which earns it more reward, while the peers it starves get no + /// traffic and so can never earn their way back. Latency has no such ratchet. + /// It expires, so a starved peer becomes unmeasured and is retried, and it is + /// bounded by what the peer actually did rather than by how often we picked it. + /// + /// The pool guard is already released by the caller (`get_all_peers` returns + /// owned `Arc`s), so taking the latency lock here keeps a single lock order. + async 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()) + let addrs: Vec = peers.iter().map(|(addr, _)| *addr).collect(); + let eligible_addrs = self.latency.lock().await.eligible(&addrs); + let eligible: Vec<&(SocketAddr, Arc>)> = + peers.iter().filter(|(addr, _)| eligible_addrs.contains(addr)).collect(); + + // `eligible` only ever narrows `peers`, and never to nothing: a peer with + // no fresh measurement always qualifies, so an empty result would mean + // every peer was measured and none was within reach of the fastest, which + // the fastest itself contradicts. + let idx = self.round_robin_counter.fetch_add(1, Ordering::Relaxed) % eligible.len(); + let (addr, peer) = eligible[idx]; + (*addr, peer.clone()) } /// Send a message to the given peer. @@ -1221,11 +1537,27 @@ impl PeerNetworkManager { 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))) + let request_kind = tracked_request_kind(&message); + let send_result = { + let mut peer_guard = peer.write().await; + peer_guard.send_message(message).await + }; + let result = send_result + .map_err(|e| NetworkError::ProtocolError(format!("Failed to send to {}: {}", addr, e))); + + // Arm the stall timer for request-type messages, keeping the oldest + // unanswered timestamp so a peer that never replies is caught. The peer + // lock is already released, so this only ever holds the outstanding lock. + if result.is_ok() { + if let Some(kind) = request_kind { + self.outstanding_requests + .lock() + .await + .entry((*addr, kind)) + .or_insert_with(Instant::now); + } + } + result } /// Broadcast a message to all connected peers @@ -1398,6 +1730,10 @@ impl Clone for PeerNetworkManager { request_rx: self.request_rx.clone(), round_robin_counter: self.round_robin_counter.clone(), network_event_sender: self.network_event_sender.clone(), + outstanding_requests: self.outstanding_requests.clone(), + latency: self.latency.clone(), + last_maintenance_at: self.last_maintenance_at.clone(), + last_eviction_at: self.last_eviction_at.clone(), } } } @@ -1488,6 +1824,81 @@ impl NetworkManager for PeerNetworkManager { } } +/// The sync request a timer belongs to. +/// +/// Timers are kept per kind rather than per peer, so a response can only clear a +/// request it actually answers. Sharing one timer across kinds lets a peer that is +/// fast at one kind hide being slow at another: its quick responses clear the +/// timer the slow request armed, the slow response then finds nothing to clear and +/// is never measured, and the stall sweep never sees an aged entry to penalize. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] +enum RequestKind { + Headers, + FilterHeaders, + Filters, + Blocks, +} + +impl RequestKind { + /// Whether how long this kind took is a fair measure of the peer, and so may + /// be scored against it: recorded as its response time, and penalized when it + /// exceeds `REQUEST_STALL_TIMEOUT`. + /// + /// Only kinds with a small, fixed-size response qualify. A header, filter or + /// filter-header reply is at most tens of KB and arrives in well under a + /// second on any usable link, so a slow one really is a slow peer. A block + /// body is megabytes and the peer sends nothing until the transfer finishes, + /// so its elapsed time measures the payload rather than the peer: scoring it + /// would both punish an honest peer on a slow link and, because response + /// times feed one shared routing metric, push that peer out of serving + /// filters and headers it was answering perfectly well. + /// + /// Blocks are still tracked, because an unanswered block request is exactly + /// what should stop us handing that peer the next one. The judgement just + /// stays at routing: the peer is skipped while it owes us, never scored. + fn response_time_is_fair(self) -> bool { + !matches!(self, RequestKind::Blocks) + } +} + +/// The kind of timer an outbound message arms, or `None` if it is not timed. +/// +/// A `getdata` only arms the block timer when it actually asks for blocks. The +/// mempool sends `getdata` for transactions, which a `block` response would never +/// clear, so timing those would leave a timer armed forever against a peer that +/// answered everything it was asked. +/// +/// Masternode-diff and quorum-info requests stay untimed: they are issued as a +/// single burst against one peer rather than as a rotation, so knowing the peer +/// owes us one changes nothing about where the next one goes. +fn tracked_request_kind(msg: &NetworkMessage) -> Option { + match msg { + NetworkMessage::GetHeaders(_) | NetworkMessage::GetHeaders2(_) => { + Some(RequestKind::Headers) + } + NetworkMessage::GetCFHeaders(_) => Some(RequestKind::FilterHeaders), + NetworkMessage::GetCFilters(_) => Some(RequestKind::Filters), + NetworkMessage::GetData(inv) => inv + .iter() + .any(|item| matches!(item, Inventory::Block(_))) + .then_some(RequestKind::Blocks), + _ => None, + } +} + +/// The kind of timer an inbound message clears, or `None` if it answers nothing we +/// timed. Unsolicited gossip (inv, tx, addr, ping) is excluded: it arrives +/// unprompted and would clear a timer the peer has not actually answered. +fn timed_response_kind(msg: &NetworkMessage) -> Option { + match msg { + NetworkMessage::Headers(_) | NetworkMessage::Headers2(_) => Some(RequestKind::Headers), + NetworkMessage::CFHeaders(_) => Some(RequestKind::FilterHeaders), + NetworkMessage::CFilter(_) => Some(RequestKind::Filters), + NetworkMessage::Block(_) => Some(RequestKind::Blocks), + _ => None, + } +} + #[cfg(test)] impl PeerNetworkManager { pub(crate) async fn new_for_test(required_services: ServiceFlags) -> Self { @@ -1519,6 +1930,10 @@ impl PeerNetworkManager { 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), + outstanding_requests: Arc::new(Mutex::new(HashMap::new())), + latency: Arc::new(Mutex::new(PeerLatency::default())), + last_maintenance_at: Arc::new(Mutex::new(Instant::now())), + last_eviction_at: Arc::new(Mutex::new(None)), } } @@ -1555,4 +1970,61 @@ impl PeerNetworkManager { pub(crate) async fn test_should_reject_after_handshake(&self, peer: &Peer) -> bool { Self::should_reject_after_handshake(&self.pool, peer, self.required_services).await } + + pub(crate) async fn test_update_reputation(&self, addr: SocketAddr, reason: ChangeReason) { + self.reputation_manager.update_reputation(addr, reason).await; + } + + pub(crate) async fn test_add_known_address(&self, addr: SocketAddr) { + self.addrv2_handler.add_known_address(addr, ServiceFlags::NETWORK).await; + } + + pub(crate) async fn test_next_peer(&self) -> SocketAddr { + let peers = self.pool.get_all_peers().await; + self.next_peer(&peers).await.0 + } + + /// Pick the peer `send_distributed` would route `msg` to, including the skip + /// of peers that still owe a response of that kind. + pub(crate) async fn test_route(&self, msg: &NetworkMessage) -> SocketAddr { + let peers = self.pool.get_all_peers().await; + let peers = match tracked_request_kind(msg) { + Some(kind) => self.without_peers_owing(kind, peers).await, + None => peers, + }; + self.next_peer(&peers).await.0 + } + + pub(crate) async fn test_record_latency(&self, addr: SocketAddr, elapsed: Duration) { + self.latency.lock().await.record(addr, elapsed); + } + + /// Arm a stall timer exactly as a successful send does. + pub(crate) async fn test_arm_request(&self, addr: SocketAddr, msg: &NetworkMessage) { + if let Some(kind) = tracked_request_kind(msg) { + self.outstanding_requests.lock().await.entry((addr, kind)).or_insert_with(Instant::now); + } + } + + /// Clear a stall timer exactly as the peer reader loop does on a response. + pub(crate) async fn test_deliver_response(&self, addr: SocketAddr, msg: &NetworkMessage) { + if let Some(kind) = timed_response_kind(msg) { + let sent = self.outstanding_requests.lock().await.remove(&(addr, kind)); + if let Some(sent) = sent { + self.latency.lock().await.record(addr, sent.elapsed()); + } + } + } + + pub(crate) async fn test_sweep_stalled_peers(&self) -> bool { + self.sweep_stalled_peers().await + } + + pub(crate) async fn test_score(&self, addr: SocketAddr) -> i32 { + self.reputation_manager.scores_for([addr]).await.get(&addr).copied().unwrap_or(0) + } + + pub(crate) async fn test_evict_worst_stuck_peer(&self) { + self.evict_worst_stuck_peer().await; + } } diff --git a/dash-spv/src/network/mod.rs b/dash-spv/src/network/mod.rs index 20c4266c7..b02adb4cd 100644 --- a/dash-spv/src/network/mod.rs +++ b/dash-spv/src/network/mod.rs @@ -5,6 +5,7 @@ pub mod constants; pub mod discovery; mod event; pub mod handshake; +mod latency; pub mod manager; mod message_dispatcher; pub mod peer; diff --git a/dash-spv/src/network/reputation.rs b/dash-spv/src/network/reputation.rs index f90656584..801f5f409 100644 --- a/dash-spv/src/network/reputation.rs +++ b/dash-spv/src/network/reputation.rs @@ -7,40 +7,40 @@ use crate::storage::PeerStorage; use dashcore::network::address::AddrV2Message; +use rand::seq::SliceRandom; use serde::{Deserialize, Deserializer, Serialize}; use std::collections::HashMap; use std::net::SocketAddr; use std::sync::Arc; -use std::time::{Duration, Instant}; +use std::time::{Duration, Instant, SystemTime}; 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. +/// Reason a peer's misbehavior score rose. Each reason owns its penalty and a +/// human-readable label. There is deliberately no reason that lowers the score: +/// only decay does that, so the score cannot be earned down by a peer that is +/// merely busy. Useful behavior is measured by [`super::latency`] instead. #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub enum ChangeReason { HandshakeFailed, ConnectionFailed, Headers2DecompressionFailed, - ReadTimeout, + RequestTimeout, PingFailed, InvalidTransactionInBlock, ManuallyBanned, - LongUptime, } impl ChangeReason { - /// Score delta for this reason: positive for misbehavior (penalty), - /// negative for good behavior (reward). + /// Penalty this reason adds to a peer's misbehavior score. pub fn score(&self) -> i32 { match self { ChangeReason::HandshakeFailed => 10, ChangeReason::ConnectionFailed => 2, ChangeReason::Headers2DecompressionFailed => 10, - ChangeReason::ReadTimeout => 5, + ChangeReason::RequestTimeout => 10, ChangeReason::PingFailed => 5, ChangeReason::InvalidTransactionInBlock => 20, ChangeReason::ManuallyBanned => 100, - ChangeReason::LongUptime => -5, } } } @@ -51,11 +51,10 @@ impl std::fmt::Display for ChangeReason { ChangeReason::HandshakeFailed => "Handshake failed", ChangeReason::ConnectionFailed => "Connection failed", ChangeReason::Headers2DecompressionFailed => "Headers2 decompression failed", - ChangeReason::ReadTimeout => "Read timeout", + ChangeReason::RequestTimeout => "Request timed out", ChangeReason::PingFailed => "Ping failed", ChangeReason::InvalidTransactionInBlock => "Invalid transaction type in block", ChangeReason::ManuallyBanned => "Manually banned", - ChangeReason::LongUptime => "Long connection uptime", }; f.write_str(label) } @@ -73,13 +72,30 @@ 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; +/// Minimum score. Zero, because this score only measures misbehavior and "none +/// observed" is as clean as a peer can be. Letting it run negative would bank +/// credit that a peer could later spend on misbehaving: decay alone would carry a +/// long-lived peer far below zero, and it would then need several extra strikes to +/// reach eviction, making the peers most likely to be quietly failing the hardest +/// ones to remove. How useful a peer actually is, as opposed to how badly it is +/// behaving, is measured by [`super::latency`] instead. +const MIN_MISBEHAVIOR_SCORE: i32 = 0; + +/// Highest score a peer may load with after a restart. A persisted near-ban +/// score can then never ban a peer on its first post-restart penalty, which +/// would otherwise let stale suspicion lock the client out of known-good peers. +const RESTART_SCORE_CEILING: i32 = 40; const MAX_BAN_COUNT: u32 = 1000; const MAX_ACTION_COUNT: u64 = 1_000_000; +/// Lowest score any earlier version could persist, back when answering a request +/// earned credit below zero. A stored score in that range is stale rather than +/// corrupt, so it is clamped quietly: warning would fire once per persisted peer +/// on the first run after an upgrade. +const LEGACY_MIN_SCORE: i32 = -50; + fn clamp_peer_score<'de, D>(deserializer: D) -> Result where D: Deserializer<'de>, @@ -87,7 +103,9 @@ where let mut v = i32::deserialize(deserializer)?; if v < MIN_MISBEHAVIOR_SCORE { - tracing::warn!("Peer has invalid score {v}, clamping to min {MIN_MISBEHAVIOR_SCORE}"); + if v < LEGACY_MIN_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}"); @@ -157,6 +175,12 @@ pub struct PeerReputation { /// Last connection time #[serde(skip)] pub last_connection: Option, + + /// Wall-clock time this reputation was last persisted. Used to credit decay + /// for time the client spent offline, since the monotonic `last_update` + /// resets on load and cannot measure downtime. + #[serde(default = "SystemTime::now")] + pub last_seen: SystemTime, } impl Default for PeerReputation { @@ -171,6 +195,7 @@ impl Default for PeerReputation { connection_attempts: 0, successful_connections: 0, last_connection: None, + last_seen: SystemTime::now(), } } } @@ -325,6 +350,22 @@ impl PeerReputationManager { reputations.clone() } + /// Return the current score for each requested peer. Peers with no record + /// default to 0. Takes a read lock and does not apply decay, so a caller sees + /// the score as of the last `update_reputation` or maintenance pass. Acquires + /// only the reputation lock, so callers must not hold a pool guard across this + /// to keep a single lock order. + pub async fn scores_for( + &self, + addrs: impl IntoIterator, + ) -> HashMap { + let reputations = self.reputations.read().await; + addrs + .into_iter() + .map(|addr| (addr, reputations.get(&addr).map_or(0, |rep| rep.score))) + .collect() + } + /// Clear banned status for a peer (admin function) pub async fn unban_peer(&self, peer: &SocketAddr) { let mut reputations = self.reputations.write().await; @@ -337,7 +378,11 @@ impl PeerReputationManager { /// 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; + let now = SystemTime::now(); + let mut reputations = self.reputations.write().await; + for reputation in reputations.values_mut() { + reputation.last_seen = now; + } storage.save_peers_reputation(&reputations).await.map_err(std::io::Error::other) } @@ -364,11 +409,21 @@ impl PeerReputationManager { 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 + // Credit decay for the wall-clock time the client spent offline. The + // monotonic `last_update` reset to now on load, so without this a peer + // never recovers while the client is closed. + if let Ok(offline) = SystemTime::now().duration_since(reputation.last_seen) { + let intervals = offline.as_secs() / DECAY_INTERVAL.as_secs(); + if intervals > 0 { + let intervals_i32 = intervals.min(i32::MAX as u64) as i32; + let decay = intervals_i32.saturating_mul(DECAY_AMOUNT); + reputation.score = (reputation.score - decay).max(MIN_MISBEHAVIOR_SCORE); + } } + // Never let a peer return one strike short of a ban after a restart. + reputation.score = reputation.score.min(RESTART_SCORE_CEILING); + reputations.insert(addr, reputation); loaded_count += 1; } @@ -421,6 +476,10 @@ impl ReputationAware for PeerReputationManager { } } + // Shuffle before the stable sort so equal-scored candidates (e.g. every + // peer at score 0 on a fresh install) are returned in random order + // rather than a fixed address order that herds clients onto the same nodes. + peer_scores.shuffle(&mut rand::thread_rng()); // Sort by score (lower is better) peer_scores.sort_by_key(|(_, score)| *score); diff --git a/dash-spv/src/network/reputation_tests.rs b/dash-spv/src/network/reputation_tests.rs index 68b74e13b..71546cab1 100644 --- a/dash-spv/src/network/reputation_tests.rs +++ b/dash-spv/src/network/reputation_tests.rs @@ -2,10 +2,12 @@ #[cfg(test)] mod tests { - use crate::storage::{PersistentPeerStorage, PersistentStorage}; + use crate::storage::{PeerStorage, PersistentPeerStorage, PersistentStorage}; use super::super::*; + use std::collections::HashMap; use std::net::SocketAddr; + use std::time::{Duration, SystemTime}; async fn score(manager: &PeerReputationManager, peer: &SocketAddr) -> i32 { manager.get_all_reputations().await.get(peer).map_or(0, |rep| rep.score) @@ -21,8 +23,8 @@ mod tests { manager.update_reputation(peer, ChangeReason::HandshakeFailed).await; assert_eq!(score(&manager, &peer).await, 10); - manager.update_reputation(peer, ChangeReason::LongUptime).await; - assert_eq!(score(&manager, &peer).await, 5); + manager.update_reputation(peer, ChangeReason::PingFailed).await; + assert_eq!(score(&manager, &peer).await, 15); } #[tokio::test] @@ -49,8 +51,7 @@ mod tests { 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(peer1, ChangeReason::PingFailed).await; manager.update_reputation(peer2, ChangeReason::InvalidTransactionInBlock).await; let temp_dir = tempfile::TempDir::new().unwrap(); @@ -62,7 +63,7 @@ mod tests { let new_manager = PeerReputationManager::new(); new_manager.load_from_storage(&peer_storage).await.unwrap(); - assert_eq!(score(&new_manager, &peer1).await, -10); + assert_eq!(score(&new_manager, &peer1).await, 5); assert_eq!(score(&new_manager, &peer2).await, 20); } @@ -70,11 +71,10 @@ mod tests { async fn test_peer_selection() { let manager = PeerReputationManager::new(); - 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 clean_peer = AddrV2Message::dummy(0, "1.1.1.1".parse().unwrap(), 8333); + let other_clean_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.update_reputation(good_peer.socket_addr().unwrap(), ChangeReason::LongUptime).await; manager .update_reputation( bad_peer.socket_addr().unwrap(), @@ -82,12 +82,15 @@ mod tests { ) .await; - let all_peers = vec![good_peer.clone(), neutral_peer.clone(), bad_peer.clone()]; + let all_peers = vec![clean_peer.clone(), other_clean_peer.clone(), bad_peer.clone()]; let selected = manager.select_best_peers(all_peers, 2).await; + // The two clean peers are indistinguishable, and deliberately so: the score + // only records misbehavior, and equal scores are returned shuffled so + // clients do not all herd onto the same node. Only their exclusion of the + // misbehaving peer is guaranteed, not their order. assert_eq!(selected.len(), 2); - assert_eq!(selected[0], good_peer.socket_addr().unwrap()); - assert_eq!(selected[1], neutral_peer.socket_addr().unwrap()); + assert!(!selected.contains(&bad_peer.socket_addr().unwrap())); } #[tokio::test] @@ -106,4 +109,72 @@ mod tests { assert_eq!(rep.connection_attempts, 2); assert_eq!(rep.successful_connections, 1); } + + #[tokio::test] + async fn test_request_timeout_penalty_and_ban() { + let manager = PeerReputationManager::new(); + let peer: SocketAddr = "127.0.0.1:7777".parse().unwrap(); + + // Ten stalls (10 * 10) reach the ban line; only the tenth bans. + for i in 0..10 { + let banned = manager.update_reputation(peer, ChangeReason::RequestTimeout).await; + assert_eq!(banned, i == 9); + } + assert_eq!(score(&manager, &peer).await, 100); + assert!(manager.is_banned(&peer).await); + } + + #[tokio::test] + async fn test_scores_for_returns_scores_and_defaults() { + let manager = PeerReputationManager::new(); + let slight: SocketAddr = "127.0.0.1:1".parse().unwrap(); + let bad: SocketAddr = "127.0.0.1:2".parse().unwrap(); + let unknown: SocketAddr = "127.0.0.1:3".parse().unwrap(); + + manager.update_reputation(slight, ChangeReason::PingFailed).await; + manager.update_reputation(bad, ChangeReason::InvalidTransactionInBlock).await; + + let scores = manager.scores_for([slight, bad, unknown]).await; + assert_eq!(scores.get(&slight), Some(&5)); + assert_eq!(scores.get(&bad), Some(&20)); + assert_eq!(scores.get(&unknown), Some(&0)); + } + + #[tokio::test] + async fn test_load_clamps_score_and_credits_offline_decay() { + let temp_dir = tempfile::TempDir::new().unwrap(); + let storage = PersistentPeerStorage::open(temp_dir.path()).await.unwrap(); + + // Persisted near the ban line but seen recently: clamps to the restart ceiling. + let near_ban: SocketAddr = "10.0.0.1:8333".parse().unwrap(); + // Persisted mid-score but offline ~2 hours: decays by two intervals on load. + let offline: SocketAddr = "10.0.0.2:8333".parse().unwrap(); + + let mut map: HashMap = HashMap::new(); + map.insert( + near_ban, + PeerReputation { + score: 95, + last_seen: SystemTime::now(), + ..Default::default() + }, + ); + map.insert( + offline, + PeerReputation { + score: 30, + last_seen: SystemTime::now() - Duration::from_secs(2 * 60 * 60 + 60), + ..Default::default() + }, + ); + storage.save_peers_reputation(&map).await.unwrap(); + + let manager = PeerReputationManager::new(); + manager.load_from_storage(&storage).await.unwrap(); + + // Clamped to RESTART_SCORE_CEILING, so one post-restart penalty cannot ban it. + assert_eq!(score(&manager, &near_ban).await, 40); + // 30 - 2 intervals * 5 decay. + assert_eq!(score(&manager, &offline).await, 20); + } } diff --git a/dash-spv/src/network/tests.rs b/dash-spv/src/network/tests.rs index 4bb3447e0..f6153bd75 100644 --- a/dash-spv/src/network/tests.rs +++ b/dash-spv/src/network/tests.rs @@ -117,3 +117,362 @@ mod pool_tests { assert_eq!(manager.test_capability_rejected_count().await, 1); } } + +#[cfg(test)] +mod selection_tests { + use crate::network::manager::{ + PeerNetworkManager, REQUEST_OWED_TIMEOUT, REQUEST_STALL_TIMEOUT, + }; + use crate::network::reputation::ChangeReason; + use crate::test_utils::test_socket_address; + use dashcore::blockdata::block::{Block, Header, Version}; + use dashcore::network::constants::ServiceFlags; + use dashcore::network::message::NetworkMessage; + use dashcore::network::message_blockdata::Inventory; + use dashcore::network::message_filter::{CFilter, GetCFHeaders, GetCFilters}; + use dashcore::{BlockHash, CompactTarget, TxMerkleNode, Txid}; + use dashcore_hashes::Hash; + use std::collections::HashMap; + use std::net::SocketAddr; + use std::time::Duration; + + fn get_filter_headers() -> NetworkMessage { + NetworkMessage::GetCFHeaders(GetCFHeaders { + filter_type: 0, + start_height: 0, + stop_hash: BlockHash::all_zeros(), + }) + } + + fn get_filters() -> NetworkMessage { + NetworkMessage::GetCFilters(GetCFilters { + filter_type: 0, + start_height: 0, + stop_hash: BlockHash::all_zeros(), + }) + } + + fn filter_response() -> NetworkMessage { + NetworkMessage::CFilter(CFilter { + filter_type: 0, + block_hash: BlockHash::all_zeros(), + filter: vec![], + }) + } + + fn get_blocks() -> NetworkMessage { + NetworkMessage::GetData(vec![Inventory::Block(BlockHash::all_zeros())]) + } + + fn get_transactions() -> NetworkMessage { + NetworkMessage::GetData(vec![Inventory::Transaction(Txid::all_zeros())]) + } + + fn block_response() -> NetworkMessage { + NetworkMessage::Block(Block { + header: Header { + version: Version::ONE, + prev_blockhash: BlockHash::all_zeros(), + merkle_root: TxMerkleNode::all_zeros(), + time: 0, + bits: CompactTarget::from_consensus(0), + nonce: 0, + }, + txdata: vec![], + }) + } + + async fn full_pool_with_bad_peer(bad: u8) -> (PeerNetworkManager, std::net::SocketAddr) { + let cf = ServiceFlags::COMPACT_FILTERS; + let manager = PeerNetworkManager::new_for_test(cf).await; + for i in 1u8..=8 { + manager.insert_test_peer(test_socket_address(i), cf).await; + } + let bad_addr = test_socket_address(bad); + // Two consecutive stalls reach the eviction threshold. + manager.test_update_reputation(bad_addr, ChangeReason::RequestTimeout).await; + manager.test_update_reputation(bad_addr, ChangeReason::RequestTimeout).await; + (manager, bad_addr) + } + + /// A peer that is fast at one request kind must not be able to hide being slow + /// at another. + /// + /// Seen on mainnet: a peer averaging 11.9s on filter headers, with a 32s p90, + /// measured as 52ms and took 40% of the traffic with zero stall penalties. Its + /// quick `cfilter` responses kept clearing the timer its slow `getcfheaders` + /// had armed, so the slow response found nothing to clear and was never + /// recorded, and the sweep never saw an aged entry. Only 255 of 1050 filter + /// header timers were cleared by an actual `cfheaders` response. + #[tokio::test(start_paused = true)] + async fn test_fast_response_does_not_clear_another_kinds_timer() { + let cf = ServiceFlags::COMPACT_FILTERS; + let manager = PeerNetworkManager::new_for_test(cf).await; + let peer = test_socket_address(1); + manager.insert_test_peer(peer, cf).await; + + // The peer owes us both a filter-header and a filter response. + manager.test_arm_request(peer, &get_filter_headers()).await; + manager.test_arm_request(peer, &get_filters()).await; + + // It answers the filter promptly. That must not count as answering the + // filter headers, which it is still sitting on. + tokio::time::advance(Duration::from_millis(50)).await; + manager.test_deliver_response(peer, &filter_response()).await; + + // Past the stall timeout with the filter headers still unanswered. + tokio::time::advance(REQUEST_STALL_TIMEOUT + Duration::from_secs(1)).await; + manager.test_sweep_stalled_peers().await; + + assert_eq!( + manager.test_score(peer).await, + ChangeReason::RequestTimeout.score(), + "the unanswered filter-header request must still be caught as a stall" + ); + } + + /// A peer stalling on several request kinds at once is one failing peer, not + /// several, so a single sweep must not stack strikes and evict it outright. + #[tokio::test(start_paused = true)] + async fn test_stalling_on_two_kinds_is_one_strike_per_sweep() { + let cf = ServiceFlags::COMPACT_FILTERS; + let manager = PeerNetworkManager::new_for_test(cf).await; + let peer = test_socket_address(1); + manager.insert_test_peer(peer, cf).await; + + manager.test_arm_request(peer, &get_filter_headers()).await; + manager.test_arm_request(peer, &get_filters()).await; + + tokio::time::advance(REQUEST_STALL_TIMEOUT + Duration::from_secs(1)).await; + manager.test_sweep_stalled_peers().await; + + assert_eq!( + manager.test_score(peer).await, + ChangeReason::RequestTimeout.score(), + "two stalled kinds are still one failing peer, so one strike" + ); + } + + #[tokio::test] + async fn test_next_peer_excludes_slow_peer() { + let cf = ServiceFlags::COMPACT_FILTERS; + let manager = PeerNetworkManager::new_for_test(cf).await; + let good1 = test_socket_address(1); + let good2 = test_socket_address(2); + let slow = test_socket_address(3); + for addr in [good1, good2, slow] { + manager.insert_test_peer(addr, cf).await; + } + + manager.test_record_latency(good1, Duration::from_millis(40)).await; + manager.test_record_latency(good2, Duration::from_millis(60)).await; + manager.test_record_latency(slow, Duration::from_secs(10)).await; + + let mut seen_good1 = false; + let mut seen_good2 = false; + for _ in 0..20 { + let picked = manager.test_next_peer().await; + assert_ne!(picked, slow, "a peer that has stopped answering must not be routed to"); + seen_good1 |= picked == good1; + seen_good2 |= picked == good2; + } + assert!(seen_good1 && seen_good2, "load should spread across the responsive peers"); + } + + /// The failure this guards against was seen on mainnet: one peer answered a + /// few requests before the others had answered any, that head start was enough + /// to rank it alone at the top, and from then on it took every request while + /// the peers it starved got none and so could never earn their way back. Here + /// the same head start must not stop the pool sharing the work. + #[tokio::test] + async fn test_next_peer_keeps_spreading_after_one_peer_gets_a_head_start() { + let cf = ServiceFlags::COMPACT_FILTERS; + let manager = PeerNetworkManager::new_for_test(cf).await; + let peers = [test_socket_address(1), test_socket_address(2), test_socket_address(3)]; + for addr in peers { + manager.insert_test_peer(addr, cf).await; + } + + // The head start: the first peer is measured, and fast, before the rest + // have answered anything at all. + for _ in 0..3 { + manager.test_record_latency(peers[0], Duration::from_millis(20)).await; + } + + let mut picks: HashMap = HashMap::new(); + for _ in 0..300 { + let picked = manager.test_next_peer().await; + *picks.entry(picked).or_default() += 1; + manager.test_record_latency(picked, Duration::from_millis(50)).await; + } + + for addr in peers { + assert_eq!( + picks.get(&addr).copied().unwrap_or(0), + 100, + "every peer should keep an equal share, got {picks:?}" + ); + } + } + + /// A peer sitting on a block request must not be handed the retry of that + /// same request. + /// + /// Seen on mainnet: one of three peers silently dropped every `getdata` it + /// received. Latency only ever records a response, so a peer answering + /// nothing was never measured and kept its turn in the rotation, which it + /// then used to win the retry of the request it had just dropped. Filter sync + /// commits behind the missing block, so the client spent 66% of a 16 minute + /// run frozen, in stalls of exactly one, two and four block timeouts. + #[tokio::test(start_paused = true)] + async fn test_peer_owing_a_block_is_skipped_until_it_answers() { + let cf = ServiceFlags::COMPACT_FILTERS; + let manager = PeerNetworkManager::new_for_test(cf).await; + let silent = test_socket_address(1); + let good = test_socket_address(2); + for addr in [silent, good] { + manager.insert_test_peer(addr, cf).await; + } + + manager.test_arm_request(silent, &get_blocks()).await; + tokio::time::advance(REQUEST_OWED_TIMEOUT + Duration::from_secs(1)).await; + + for _ in 0..10 { + assert_eq!( + manager.test_route(&get_blocks()).await, + good, + "a peer already sitting on a block request must not be sent another" + ); + } + + // Answering clears the debt, and the peer is a candidate again. + manager.test_deliver_response(silent, &block_response()).await; + let mut picks: HashMap = HashMap::new(); + for _ in 0..10 { + *picks.entry(manager.test_route(&get_blocks()).await).or_default() += 1; + } + assert!(picks.contains_key(&silent), "a peer that answered must be routed to again"); + } + + /// The skip is a preference, not a rule: with every peer owing a response, + /// refusing to send is worse than sending to a busy peer. + #[tokio::test(start_paused = true)] + async fn test_routing_falls_back_when_every_peer_owes_a_block() { + let cf = ServiceFlags::COMPACT_FILTERS; + let manager = PeerNetworkManager::new_for_test(cf).await; + let peers = [test_socket_address(1), test_socket_address(2)]; + for addr in peers { + manager.insert_test_peer(addr, cf).await; + manager.test_arm_request(addr, &get_blocks()).await; + } + tokio::time::advance(REQUEST_OWED_TIMEOUT + Duration::from_secs(1)).await; + + assert!(peers.contains(&manager.test_route(&get_blocks()).await)); + } + + /// Blocks are routed away from, never penalized for. A block body can be + /// megabytes and a peer sends nothing until the transfer completes, so a slow + /// link is indistinguishable from a peer ignoring us and must not cost + /// reputation or trigger eviction. + #[tokio::test(start_paused = true)] + async fn test_unanswered_block_request_does_not_penalize_the_peer() { + let cf = ServiceFlags::COMPACT_FILTERS; + let manager = PeerNetworkManager::new_for_test(cf).await; + let peer = test_socket_address(1); + manager.insert_test_peer(peer, cf).await; + + manager.test_arm_request(peer, &get_blocks()).await; + tokio::time::advance(REQUEST_STALL_TIMEOUT + Duration::from_secs(1)).await; + manager.test_sweep_stalled_peers().await; + + assert_eq!( + manager.test_score(peer).await, + 0, + "a slow block transfer must not be scored as a stall" + ); + } + + /// The mempool asks for transactions with the same `getdata` message blocks + /// use, and a `block` response would never arrive to clear that timer. Timing + /// it would leave a peer permanently owing a response it was never asked for. + #[tokio::test(start_paused = true)] + async fn test_transaction_getdata_does_not_arm_the_block_timer() { + let cf = ServiceFlags::COMPACT_FILTERS; + let manager = PeerNetworkManager::new_for_test(cf).await; + // Only one peer is sent the tx request, so a wrongly armed block timer + // shows up as that peer being skipped rather than being hidden by the + // all-peers-owing fallback. + let mempool_peer = test_socket_address(1); + let idle_peer = test_socket_address(2); + for addr in [mempool_peer, idle_peer] { + manager.insert_test_peer(addr, cf).await; + } + manager.test_arm_request(mempool_peer, &get_transactions()).await; + tokio::time::advance(REQUEST_OWED_TIMEOUT + Duration::from_secs(1)).await; + + let mut picks: HashMap = HashMap::new(); + for _ in 0..20 { + *picks.entry(manager.test_route(&get_blocks()).await).or_default() += 1; + } + assert!( + picks.contains_key(&mempool_peer), + "a pending tx request must not make a peer look like it owes a block, got {picks:?}" + ); + } + + #[tokio::test] + async fn test_evict_worst_stuck_peer_removes_the_stalling_peer() { + let (manager, bad) = full_pool_with_bad_peer(3).await; + // A replacement must be available or eviction is a no-op. + manager.test_add_known_address(test_socket_address(99)).await; + + manager.test_evict_worst_stuck_peer().await; + + assert!(!manager.test_is_connected(&bad).await, "stalling peer should be evicted"); + assert_eq!(manager.test_peer_count().await, 7); + } + + #[tokio::test] + async fn test_evict_worst_stuck_peer_skips_without_replacement() { + let (manager, _bad) = full_pool_with_bad_peer(3).await; + // No known addresses -> no replacement candidate -> no eviction. + manager.test_evict_worst_stuck_peer().await; + assert_eq!(manager.test_peer_count().await, 8); + } + + #[tokio::test] + async fn test_evict_worst_stuck_peer_skips_when_all_peers_bad() { + let cf = ServiceFlags::COMPACT_FILTERS; + let manager = PeerNetworkManager::new_for_test(cf).await; + for i in 1u8..=8 { + let addr = test_socket_address(i); + manager.insert_test_peer(addr, cf).await; + manager.test_update_reputation(addr, ChangeReason::RequestTimeout).await; + manager.test_update_reputation(addr, ChangeReason::RequestTimeout).await; + } + manager.test_add_known_address(test_socket_address(99)).await; + + manager.test_evict_worst_stuck_peer().await; + + // Every peer is equally bad, so the problem is systemic: evict none. + assert_eq!(manager.test_peer_count().await, 8); + } + + #[tokio::test] + async fn test_evict_worst_stuck_peer_skips_when_pool_not_full() { + let cf = ServiceFlags::COMPACT_FILTERS; + let manager = PeerNetworkManager::new_for_test(cf).await; + for i in 1u8..=3 { + manager.insert_test_peer(test_socket_address(i), cf).await; + } + let bad = test_socket_address(2); + manager.test_update_reputation(bad, ChangeReason::RequestTimeout).await; + manager.test_update_reputation(bad, ChangeReason::RequestTimeout).await; + manager.test_add_known_address(test_socket_address(99)).await; + + manager.test_evict_worst_stuck_peer().await; + + // Pool is below max_peers, so eviction would be pure loss: skip. + assert_eq!(manager.test_peer_count().await, 3); + } +} diff --git a/dash-spv/src/sync/blocks/pipeline.rs b/dash-spv/src/sync/blocks/pipeline.rs index 02b29aac2..8610d33b8 100644 --- a/dash-spv/src/sync/blocks/pipeline.rs +++ b/dash-spv/src/sync/blocks/pipeline.rs @@ -17,7 +17,12 @@ use key_wallet_manager::{FilterMatchKey, WalletId}; const MAX_CONCURRENT_BLOCK_DOWNLOADS: usize = 20; /// Timeout for block downloads before retry. -const BLOCK_TIMEOUT: Duration = Duration::from_secs(30); +/// +/// A filter batch cannot commit while any block it matched is outstanding, so +/// this is the price of a single peer dropping a request, paid with filter sync +/// stopped. It stays comfortably above the transfer time of a full block on a +/// slow link, but no longer treats a peer that will never answer as merely slow. +const BLOCK_TIMEOUT: Duration = Duration::from_secs(15); /// Maximum blocks per GetData request, kept a bit lower for better download distribution to multiple peers const BLOCKS_PER_REQUEST: usize = 8; diff --git a/dash-spv/src/sync/filters/manager.rs b/dash-spv/src/sync/filters/manager.rs index f01fbcb59..4bfffb27c 100644 --- a/dash-spv/src/sync/filters/manager.rs +++ b/dash-spv/src/sync/filters/manager.rs @@ -43,6 +43,13 @@ struct WalletScanState { } /// Maximum number of batches to scan ahead while waiting for blocks. +/// +/// Raising this does not buy more throughput while a batch is stalled on a block. +/// Every scanned batch queues its matched blocks into one FIFO download window, +/// so a deeper lookahead fills that window with blocks belonging to batches that +/// cannot commit for a long time, ahead of the blocks the committing batch is +/// actually waiting on. Block downloads would have to be ordered by height +/// before this could go higher. const MAX_LOOKAHEAD_BATCHES: usize = 3; /// Filters manager for downloading and matching compact block filters. @@ -1405,10 +1412,36 @@ mod tests { assert_eq!(manager.progress.filter_header_tip_height(), 600); } + /// Lookahead is what lets scanning continue past a batch that cannot commit + /// because a block it matched is still downloading, so it must actually reach + /// the cap rather than stopping at whatever the first pass created. #[tokio::test] - async fn test_max_lookahead_constant() { - // Verify the constant is set to expected value - assert_eq!(MAX_LOOKAHEAD_BATCHES, 3); + async fn test_lookahead_fills_to_the_cap_and_stops() { + let mut manager = create_test_manager().await; + manager.set_state(SyncState::Syncing); + + // Far more headroom than the cap, so the cap is what stops it. The + // filters themselves are still downloading, which is the state lookahead + // exists for: the batches are created empty and scanned once they land. + let tip = BATCH_PROCESSING_SIZE * (MAX_LOOKAHEAD_BATCHES as u32 + 4); + manager.processing_height = 1; + manager.progress.update_filter_header_tip_height(tip); + manager.progress.update_target_height(tip); + + manager.try_create_lookahead_batches().await.unwrap(); + + assert_eq!(manager.active_batches.len(), MAX_LOOKAHEAD_BATCHES); + + // A second pass adds nothing while the batches are still uncommitted. + manager.try_create_lookahead_batches().await.unwrap(); + assert_eq!(manager.active_batches.len(), MAX_LOOKAHEAD_BATCHES); + + // The batches must tile the range contiguously from the processing head. + let mut expected_start = manager.processing_height; + for (&start, batch) in &manager.active_batches { + assert_eq!(start, expected_start); + expected_start = batch.end_height() + 1; + } } #[tokio::test]