hashiverse_lib/client/peer_tracker/
peer_tracker.rs

1//! # Kademlia-style local peer set
2//!
3//! `PeerTracker` is the client's working memory of the network: a deduplicated list of
4//! [`crate::protocol::peer::Peer`] records loaded from `BUCKET_PEER`, validated, and
5//! indexed by XOR distance so [`crate::client::peer_tracker::peer_iterator::PeerIterator`]
6//! can walk them in order of closeness to any target
7//! [`crate::tools::types::Id`].
8//!
9//! Responsibilities:
10//!
11//! - **Bootstrap** — when the tracker is empty (new install, wiped storage) it calls
12//!   `BootstrapV1` against seed domains from [`crate::tools::config::BOOTSTRAP_DOMAINS`]
13//!   to obtain a starting set of peers.
14//! - **Freshness** — stale peers (missed announces, failed RPCs) are demoted or dropped;
15//!   fresh ones from gossip (`AnnounceV2`) or from RPC responses are folded in.
16//! - **Flush** — peer mutations set `peers_need_flush`; the outer client batches those
17//!   into periodic writes to storage so every RPC response doesn't trigger a disk write.
18//!
19//! ## Locking
20//!
21//! The tracker is shared as a [`SharedPeerTracker`] (`Arc<parking_lot::RwLock<PeerTracker>>`) so that
22//! concurrent DHT walks can interleave.  Every critical section is synchronous and short: anything that
23//! awaits (RPCs, storage writes) or is expensive (peer verification) happens with no guard held.
24//! `parking_lot` guards are `!Send`, so holding one across an `.await` in a `Send` future fails to compile.
25
26use crate::anyhow_assert_eq;
27use crate::client::client_storage::client_storage;
28use crate::client::client_storage::client_storage::{ClientStorage, BUCKET_PEER};
29use crate::client::peer_tracker::peer_iterator::PeerIterator;
30use crate::protocol::payload::payload::{BootstrapResponseV1, BootstrapV1, PayloadRequestKind, PayloadResponseKind};
31use crate::protocol::peer::Peer;
32use crate::protocol::rpc;
33use crate::tools::runtime_services::RuntimeServices;
34use crate::tools::time::TimeMillis;
35use crate::tools::tools::LeadingAgreementBits;
36use crate::tools::types::Id;
37use crate::tools::{config, json, tools};
38use log::{info, trace, warn};
39use std::sync::Arc;
40
41/// The client-wide handle to the peer tracker.  Lock it only for short synchronous sections - never across an `.await`.
42pub type SharedPeerTracker = Arc<parking_lot::RwLock<PeerTracker>>;
43
44/// The client's local view of the known peer set and the entry point for Kademlia-style
45/// routing.
46///
47/// `PeerTracker` owns the in-memory list of [`Peer`] records the client has seen, persists
48/// them to [`ClientStorage`] under `BUCKET_PEER` so they survive restarts, and exposes the
49/// iteration primitives used throughout the client when it needs to answer "who should I
50/// talk to next about this [`Id`]?". When the list is empty (first launch, or after a
51/// reset), it seeds itself via a `BootstrapV1` RPC against the
52/// [`crate::transport::bootstrap_provider::bootstrap_provider::BootstrapProvider`] addresses configured on the
53/// transport.
54///
55/// The tracker is the single source of truth for peer freshness: stale or bad peers get
56/// evicted here, new peers get folded in here, and the `peers_need_flush` flag coalesces
57/// rapid updates into a single disk write.
58pub struct PeerTracker {
59    runtime_services: Arc<RuntimeServices>,
60    client_storage: Arc<dyn ClientStorage>,
61    peers_need_flush: bool,
62    peers: Vec<Peer>,
63    max_peers: usize,
64    /// Serialises flushes so a slow, older snapshot can never land in storage after a newer one.
65    flush_serialiser: Arc<tokio::sync::Mutex<()>>,
66}
67
68impl PeerTracker {
69    pub async fn new(runtime_services: Arc<RuntimeServices>, client_storage: Arc<dyn ClientStorage>) -> anyhow::Result<Self> {
70        Self::new_with_max_peers(runtime_services, client_storage, config::CLIENT_PEER_TRACKER_MAX_PEERS).await
71    }
72
73    pub async fn new_with_max_peers(runtime_services: Arc<RuntimeServices>, client_storage: Arc<dyn ClientStorage>, max_peers: usize) -> anyhow::Result<Self> {
74        let peers: anyhow::Result<Vec<Peer>> = try {
75            let peers = client_storage::get_struct::<Vec<Peer>>(client_storage.as_ref(), BUCKET_PEER, "peers", runtime_services.time_provider.current_time_millis()).await?;
76            match peers {
77                Some(peers) => {
78                    info!("PeerTracker is starting with {} peers", peers.len());
79                    trace!("{:?}", peers);
80                    peers
81                }
82                None => Vec::new(),
83            }
84        };
85
86        let peers = peers.unwrap_or_else(|e| {
87            warn!("Failed to load peers from storage: {}", e);
88            Vec::new()
89        });
90
91        let current_time_millis = runtime_services.time_provider.current_time_millis();
92        let (peers, peers_were_dropped) = Self::sanitise_loaded_peers(peers, current_time_millis, max_peers);
93
94        Ok(Self {
95            runtime_services,
96            client_storage,
97            peers_need_flush: peers_were_dropped,
98            peers,
99            max_peers,
100            flush_serialiser: Arc::new(tokio::sync::Mutex::new(())),
101        })
102    }
103
104    /// Persisted peers predate the current bounds: drop far-future records (which could otherwise never be replaced,
105    /// because newest-timestamp wins) and trim to `max_peers` by score.  Returns the peers sorted by id.
106    fn sanitise_loaded_peers(peers: Vec<Peer>, current_time_millis: TimeMillis, max_peers: usize) -> (Vec<Peer>, bool) {
107        let loaded_peer_count = peers.len();
108        let mut peers: Vec<Peer> = peers
109            .into_iter()
110            .filter(|peer| match peer.verify_timestamps(current_time_millis) {
111                Ok(()) => true,
112                Err(e) => {
113                    warn!("dropping stored peer: {}", e);
114                    false
115                }
116            })
117            .collect();
118
119        if peers.len() > max_peers {
120            peers.sort_by(|a, b| b.score(current_time_millis).total_cmp(&a.score(current_time_millis)));
121            peers.truncate(max_peers);
122        }
123        peers.sort_by_key(|peer| peer.id);
124
125        let peers_were_dropped = peers.len() != loaded_peer_count;
126        (peers, peers_were_dropped)
127    }
128
129    /// Clears the dirty flag and returns the peers to persist, or `None` if nothing changed since the last flush.
130    pub fn take_flush_snapshot(&mut self) -> Option<Vec<Peer>> {
131        if !self.peers_need_flush {
132            return None;
133        }
134        self.peers_need_flush = false;
135        Some(self.peers.clone())
136    }
137
138    pub fn mark_needs_flush(&mut self) {
139        self.peers_need_flush = true;
140    }
141
142    pub fn current_time_millis(&self) -> TimeMillis {
143        self.runtime_services.time_provider.current_time_millis()
144    }
145
146    /// Verify-and-insert in one go.  Holders of a [`SharedPeerTracker`] should instead call [`Self::needs_verification`] under a read lock,
147    /// [`Self::verify_candidate`] with no lock held, then [`Self::insert_verified`] under a write lock - verification is the expensive part.
148    pub fn add_peer(&mut self, peer: Peer) -> anyhow::Result<()> {
149        if !self.needs_verification(&peer) {
150            return Ok(());
151        }
152        Self::verify_candidate(&peer, self.current_time_millis())?;
153        self.insert_verified(peer);
154        Ok(())
155    }
156
157    /// Cheap dedupe before the expensive verify: servers resend the same peers in every peers_nearer list.
158    /// False if we already hold this peer with a timestamp at least as new.
159    pub fn needs_verification(&self, peer: &Peer) -> bool {
160        match self.peers.binary_search_by_key(&peer.id, |peer| peer.id) {
161            Ok(i) => peer.timestamp > self.peers[i].timestamp,
162            Err(_) => true,
163        }
164    }
165
166    /// Everything a peer must pass before it may enter the tracker.  Pure, so it can run with no lock held.
167    pub fn verify_candidate(peer: &Peer, current_time_millis: TimeMillis) -> anyhow::Result<()> {
168        // Sanity check that this peer is kosher
169        if let Err(e) = peer.verify() {
170            anyhow::bail!("peer verification error: {}", e);
171        }
172
173        // Check that its pow is reasonable
174        if peer.pow_initial.pow < config::SERVER_KEY_POW_MIN {
175            anyhow::bail!("peer peer.pow_initial.pow={} < {}", peer.pow_initial.pow, config::SERVER_KEY_POW_MIN);
176        }
177
178        // A far-future timestamp would win every later newest-timestamp comparison, so the real peer could never replace it
179        peer.verify_timestamps(current_time_millis)?;
180
181        Ok(())
182    }
183
184    /// Inserts a peer that has passed [`Self::verify_candidate`].  Re-checks the timestamp dedupe, because another walk may have
185    /// inserted a newer copy between our verify and this insert.
186    pub fn insert_verified(&mut self, peer: Peer) {
187        match self.peers.binary_search_by_key(&peer.id, |peer| peer.id) {
188            Ok(i) => {
189                if peer.timestamp <= self.peers[i].timestamp {
190                    return;
191                }
192                self.peers[i] = peer;
193            }
194            Err(i) => {
195                if self.peers.len() < self.max_peers {
196                    self.peers.insert(i, peer);
197                }
198                else {
199                    // Full: the newcomer only gets in by out-scoring our weakest peer
200                    let current_time_millis = self.current_time_millis();
201                    let weakest_peer = self.peers.iter().enumerate().map(|(index, existing_peer)| (index, existing_peer.score(current_time_millis))).min_by(|a, b| a.1.total_cmp(&b.1));
202                    let Some((weakest_peer_index, weakest_peer_score)) = weakest_peer else { return };
203                    if peer.score(current_time_millis) <= weakest_peer_score {
204                        return;
205                    }
206                    self.peers.remove(weakest_peer_index);
207                    let insert_index = self.peers.binary_search_by_key(&peer.id, |peer| peer.id).unwrap_or_else(|insert_index| insert_index);
208                    self.peers.insert(insert_index, peer);
209                }
210            }
211        }
212
213        self.peers_need_flush = true;
214    }
215
216    pub fn remove_peer(&mut self, peer: &Peer) {
217        if let Ok(i) = self.peers.binary_search_by_key(&peer.id, |peer| peer.id) {
218            self.peers.remove(i);
219            self.peers_need_flush = true;
220        }
221    }
222
223    pub fn is_empty(&self) -> bool {
224        self.peers.is_empty()
225    }
226    pub fn needs_flush(&self) -> bool {
227        self.peers_need_flush
228    }
229    pub fn len(&self) -> usize {
230        self.peers.len()
231    }
232
233    pub fn peers(&self) -> &Vec<Peer> {
234        &self.peers
235    }
236
237    /// Persists the peers if they changed.  The storage write happens with no tracker lock held.
238    pub async fn flush(peer_tracker: &SharedPeerTracker) -> anyhow::Result<()> {
239        let flush_serialiser = peer_tracker.read().flush_serialiser.clone();
240        let _flush_guard = flush_serialiser.lock().await;
241
242        let (client_storage, current_time_millis, peers) = {
243            let mut peer_tracker = peer_tracker.write();
244            let Some(peers) = peer_tracker.take_flush_snapshot() else { return Ok(()) };
245            (peer_tracker.client_storage.clone(), peer_tracker.current_time_millis(), peers)
246        };
247
248        trace!("Flushing peers to storage");
249        let result = client_storage::put_struct(client_storage.as_ref(), BUCKET_PEER, "peers", &peers, current_time_millis).await;
250        if result.is_err() {
251            peer_tracker.write().mark_needs_flush();
252        }
253        result
254    }
255
256    /// Bootstraps if necessary, then starts a walk towards `bucket_location_id`.  The iterator locks the tracker only briefly per step.
257    pub async fn iterate_to_location(peer_tracker: &SharedPeerTracker, bucket_location_id: Id, max_iterations_since_high_watermark: usize, cache_radius: Option<LeadingAgreementBits>) -> anyhow::Result<PeerIterator> {
258        Self::bootstrap(peer_tracker).await?;
259
260        Ok(PeerIterator::new(peer_tracker.clone(), bucket_location_id, max_iterations_since_high_watermark, cache_radius))
261    }
262
263    /// Seeds an empty tracker from the bootstrap addresses.  RPCs and verification happen with no lock held; two concurrent
264    /// bootstraps may both fetch, which is harmless because inserts are idempotent.
265    pub async fn bootstrap(peer_tracker: &SharedPeerTracker) -> anyhow::Result<()> {
266        // We only need to bootstrap if we have noone to talk to!
267        let runtime_services = {
268            let peer_tracker = peer_tracker.read();
269            if !peer_tracker.is_empty() {
270                return Ok(());
271            }
272            peer_tracker.runtime_services.clone()
273        };
274
275        info!("bootstrapping PeerTracker");
276
277        // Lets randomize these addresses so that the first one is not snowed
278        let mut bootstrap_addresses = runtime_services.transport_factory.get_bootstrap_addresses().await;
279        tools::shuffle(&mut bootstrap_addresses);
280
281        // Our bootstrap process has handed us a bunch of raw addresses, so we need to convert them into peers
282        for bootstrap_address in bootstrap_addresses {
283            let try_result: anyhow::Result<()> = try {
284                info!("bootstrapping {}", bootstrap_address);
285
286                let request = json::struct_to_bytes(&BootstrapV1 {})?;
287                let response = rpc::rpc::rpc_server_unknown(&runtime_services, &Id::zero(), &bootstrap_address, PayloadRequestKind::BootstrapV1, request).await?;
288                anyhow_assert_eq!(&PayloadResponseKind::BootstrapResponseV1, &response.response_request_kind);
289                let response = json::bytes_to_struct::<BootstrapResponseV1>(&response.bytes)?;
290                if response.peers_random.len() > config::CLIENT_MAX_PEERS_CONSIDERED_PER_RESPONSE {
291                    warn!("ignoring {} of {} bootstrapped peers from {} (cap is {})", response.peers_random.len() - config::CLIENT_MAX_PEERS_CONSIDERED_PER_RESPONSE, response.peers_random.len(), bootstrap_address, config::CLIENT_MAX_PEERS_CONSIDERED_PER_RESPONSE);
292                }
293
294                let current_time_millis = runtime_services.time_provider.current_time_millis();
295                let verified_peers: Vec<Peer> = response
296                    .peers_random
297                    .into_iter()
298                    .take(config::CLIENT_MAX_PEERS_CONSIDERED_PER_RESPONSE)
299                    .filter(|peer| match Self::verify_candidate(peer, current_time_millis) {
300                        Ok(()) => true,
301                        Err(e) => {
302                            warn!("problem while adding bootstrapped peer: {}", e);
303                            false
304                        }
305                    })
306                    .collect();
307
308                let mut peer_tracker = peer_tracker.write();
309                for peer in verified_peers {
310                    peer_tracker.insert_verified(peer);
311                }
312            };
313
314            if let Err(e) = try_result {
315                warn!("problem bootstrapping peer {}: {}", bootstrap_address, e);
316            }
317
318            // We only need to continue bootstrapping if we still have noone to talk to!
319            if !peer_tracker.read().is_empty() {
320                break;
321            }
322        }
323
324        Ok(())
325    }
326}
327
328#[cfg(test)]
329mod tests {
330    use crate::client::client_storage::mem_client_storage::MemClientStorage;
331    use crate::client::peer_tracker::peer_tracker::{PeerTracker, SharedPeerTracker};
332    use crate::tools::config;
333    use crate::tools::runtime_services::RuntimeServices;
334    use crate::tools::server_id::ServerId;
335    use crate::tools::types::Pow;
336
337    #[tokio::test]
338    async fn general_tests() -> anyhow::Result<()> {
339        let runtime_services = RuntimeServices::default_for_testing();
340        let client_storage = MemClientStorage::new().await?;
341        let mut peer_tracker = PeerTracker::new(runtime_services.clone(), client_storage.clone()).await?;
342
343        assert!(peer_tracker.is_empty());
344        assert_eq!(0, peer_tracker.len());
345
346        // Dont accept insufficient pow
347        {
348            loop {
349                let server_id = ServerId::new("own_pow", runtime_services.time_provider.as_ref(), Pow(config::SERVER_KEY_POW_MIN.0 / 2), true, runtime_services.pow_generator.as_ref()).await?;
350                if server_id.pow >= config::SERVER_KEY_POW_MIN {
351                    continue;
352                }
353                let peer = server_id.to_peer(runtime_services.time_provider.as_ref())?;
354                let result = peer_tracker.add_peer(peer);
355                assert!(result.is_err());
356                assert_eq!(0, peer_tracker.len());
357                break;
358            }
359        }
360
361        // Add an individual
362        {
363            let server_id = ServerId::new("own_pow", runtime_services.time_provider.as_ref(), config::SERVER_KEY_POW_MIN, true, runtime_services.pow_generator.as_ref()).await?;
364            let peer = server_id.to_peer(runtime_services.time_provider.as_ref())?;
365            let result = peer_tracker.add_peer(peer);
366            assert!(result.is_ok());
367            assert_eq!(1, peer_tracker.len());
368        }
369
370        // Cant add individual twice
371        {
372            let server_id = ServerId::new("own_pow", runtime_services.time_provider.as_ref(), config::SERVER_KEY_POW_MIN, true, runtime_services.pow_generator.as_ref()).await?;
373            let peer = server_id.to_peer(runtime_services.time_provider.as_ref())?;
374            let result = peer_tracker.add_peer(peer.clone());
375            assert!(result.is_ok());
376            assert_eq!(2, peer_tracker.len());
377            let result = peer_tracker.add_peer(peer.clone());
378            assert!(result.is_ok());
379            assert_eq!(2, peer_tracker.len());
380        }
381
382        // Add an individual, then remove it
383        {
384            let server_id = ServerId::new("own_pow", runtime_services.time_provider.as_ref(), config::SERVER_KEY_POW_MIN, true, runtime_services.pow_generator.as_ref()).await?;
385            let peer = server_id.to_peer(runtime_services.time_provider.as_ref())?;
386            let result = peer_tracker.add_peer(peer.clone());
387            assert!(result.is_ok());
388            assert_eq!(3, peer_tracker.len());
389            peer_tracker.remove_peer(&peer);
390            assert_eq!(2, peer_tracker.len());
391        }
392
393        // Remove an unknown individual
394        {
395            let server_id = ServerId::new("own_pow", runtime_services.time_provider.as_ref(), config::SERVER_KEY_POW_MIN, true, runtime_services.pow_generator.as_ref()).await?;
396            let peer = server_id.to_peer(runtime_services.time_provider.as_ref())?;
397            peer_tracker.remove_peer(&peer);
398            assert_eq!(2, peer_tracker.len());
399        }
400
401        Ok(())
402    }
403
404    /// A peer whose pow_initial has been mutated after signing must fail verify(),
405    /// so add_peer rejects it.
406    #[tokio::test]
407    async fn add_peer_rejects_tampered_pow_initial() -> anyhow::Result<()> {
408        let runtime_services = RuntimeServices::default_for_testing();
409        let client_storage = MemClientStorage::new().await?;
410        let mut peer_tracker = PeerTracker::new(runtime_services.clone(), client_storage.clone()).await?;
411
412        let server_id = ServerId::new("own_pow", runtime_services.time_provider.as_ref(), config::SERVER_KEY_POW_MIN, true, runtime_services.pow_generator.as_ref()).await?;
413        let mut peer = server_id.to_peer(runtime_services.time_provider.as_ref())?;
414
415        peer.pow_initial.salt = crate::tools::types::Salt::zero();
416
417        let result = peer_tracker.add_peer(peer);
418        assert!(result.is_err(), "tampered peer should be rejected");
419        assert_eq!(0, peer_tracker.len());
420
421        Ok(())
422    }
423
424    // ── Bounded tracker ── (ManualTimeProvider is native-only)
425    #[cfg(not(target_arch = "wasm32"))]
426    mod bounded_tracker {
427        use super::*;
428
429        use crate::client::client_storage::client_storage::{self, BUCKET_PEER};
430        use crate::client::peer_tracker::peer_iterator::PeerIterator;
431        use crate::protocol::peer::Peer;
432        use crate::tools::time::{TimeMillis, MILLIS_IN_DAY};
433        use crate::tools::time_provider::manual_time_provider::ManualTimeProvider;
434        use crate::tools::types::Id;
435        use crate::tools::pow_generator::single_threaded_pow_generator::SingleThreadedPowGenerator;
436        use crate::transport::mem_transport::MemTransportFactory;
437        use std::sync::Arc;
438
439        const START_TIME_MILLIS: TimeMillis = TimeMillis(1_800_000_000_000);
440
441        fn manual_runtime_services() -> (Arc<RuntimeServices>, Arc<ManualTimeProvider>) {
442            let manual_time_provider = Arc::new(ManualTimeProvider::new(START_TIME_MILLIS));
443            let runtime_services = Arc::new(RuntimeServices {
444                time_provider: manual_time_provider.clone(),
445                transport_factory: MemTransportFactory::default(),
446                pow_generator: Arc::new(SingleThreadedPowGenerator::new()),
447            });
448            (runtime_services, manual_time_provider)
449        }
450
451        /// A valid peer whose pows and signature are all dated "now" according to `runtime_services`.
452        async fn make_peer(runtime_services: &RuntimeServices) -> anyhow::Result<(ServerId, Peer)> {
453            let server_id = ServerId::new("own_pow", runtime_services.time_provider.as_ref(), config::SERVER_KEY_POW_MIN, true, runtime_services.pow_generator.as_ref()).await?;
454            let peer = server_id.to_peer(runtime_services.time_provider.as_ref())?;
455            Ok((server_id, peer))
456        }
457
458        fn peer_ids(peer_tracker: &PeerTracker) -> Vec<Id> {
459            peer_tracker.peers().iter().map(|peer| peer.id).collect()
460        }
461
462        #[tokio::test]
463        async fn add_peer_never_exceeds_max_peers() -> anyhow::Result<()> {
464            let runtime_services = RuntimeServices::default_for_testing();
465            let client_storage = MemClientStorage::new().await?;
466            const MAX_PEERS: usize = 4;
467            let mut peer_tracker = PeerTracker::new_with_max_peers(runtime_services.clone(), client_storage, MAX_PEERS).await?;
468
469            for _ in 0..MAX_PEERS + 4 {
470                let (_, peer) = make_peer(&runtime_services).await?;
471                peer_tracker.add_peer(peer)?;
472                assert!(peer_tracker.len() <= MAX_PEERS);
473            }
474            assert_eq!(MAX_PEERS, peer_tracker.len());
475            for window in peer_tracker.peers().windows(2) {
476                assert!(window[0].id < window[1].id, "peers must stay sorted by id");
477            }
478            Ok(())
479        }
480
481        #[tokio::test]
482        async fn full_tracker_evicts_weakest_peer_only_for_a_stronger_newcomer() -> anyhow::Result<()> {
483            let (runtime_services, manual_time_provider) = manual_runtime_services();
484            let client_storage = MemClientStorage::new().await?;
485            let mut peer_tracker = PeerTracker::new_with_max_peers(runtime_services.clone(), client_storage, 2).await?;
486
487            // Two peers whose pows are 90 days old by the time the newcomer arrives: their score has decayed away.
488            let (_, stale_peer_1) = make_peer(&runtime_services).await?;
489            let (_, stale_peer_2) = make_peer(&runtime_services).await?;
490            // A peer that will be fresh at the end, created after the clock moves on.
491            manual_time_provider.set_time(START_TIME_MILLIS + MILLIS_IN_DAY.const_mul(90));
492            let (_, fresh_peer) = make_peer(&runtime_services).await?;
493            let (_, other_fresh_peer) = make_peer(&runtime_services).await?;
494            let now = runtime_services.time_provider.current_time_millis();
495            let weakest_fresh_score = fresh_peer.score(now).min(other_fresh_peer.score(now));
496            let strongest_stale_score = stale_peer_1.score(now).max(stale_peer_2.score(now));
497            assert!(weakest_fresh_score > strongest_stale_score, "test setup: fresh peers must out-score stale peers");
498
499            // Full with {stale, fresh}: a stronger newcomer evicts the stale peer and keeps the fresh one.
500            peer_tracker.add_peer(stale_peer_1.clone())?;
501            peer_tracker.add_peer(fresh_peer.clone())?;
502            peer_tracker.add_peer(other_fresh_peer.clone())?;
503            let mut expected_ids = vec![fresh_peer.id, other_fresh_peer.id];
504            expected_ids.sort();
505            assert_eq!(expected_ids, peer_ids(&peer_tracker), "the weakest peer must be evicted for a stronger newcomer");
506
507            // Full with {fresh, fresh}: a weaker newcomer is dropped and nothing changes.
508            peer_tracker.add_peer(stale_peer_2.clone())?;
509            assert_eq!(expected_ids, peer_ids(&peer_tracker), "a weaker newcomer must be dropped");
510            Ok(())
511        }
512
513        #[tokio::test]
514        async fn add_peer_rejects_far_future_timestamp() -> anyhow::Result<()> {
515            let (runtime_services, _) = manual_runtime_services();
516            let client_storage = MemClientStorage::new().await?;
517            let mut peer_tracker = PeerTracker::new(runtime_services.clone(), client_storage).await?;
518
519            let (server_id, mut peer) = make_peer(&runtime_services).await?;
520            let future_time_provider = ManualTimeProvider::new(START_TIME_MILLIS + config::PEER_MAX_FUTURE_TIMESTAMP_MILLIS.const_mul(2));
521            peer.sign(&future_time_provider, &server_id.keys.signature_key)?;
522            assert!(peer.verify().is_ok(), "test setup: the record itself is validly signed");
523
524            assert!(peer_tracker.add_peer(peer).is_err(), "far-future peer must be rejected");
525            assert_eq!(0, peer_tracker.len());
526            Ok(())
527        }
528
529        #[tokio::test]
530        async fn add_peer_skips_verify_for_known_peer_that_is_not_newer() -> anyhow::Result<()> {
531            let runtime_services = RuntimeServices::default_for_testing();
532            let client_storage = MemClientStorage::new().await?;
533            let mut peer_tracker = PeerTracker::new(runtime_services.clone(), client_storage).await?;
534
535            let (_, peer) = make_peer(&runtime_services).await?;
536            peer_tracker.add_peer(peer.clone())?;
537            assert!(peer_tracker.take_flush_snapshot().is_some());
538            assert!(!peer_tracker.needs_flush());
539
540            // Same timestamp but invalid: if verify ran this would be rejected; the dedupe must short-circuit first.
541            let mut tampered_same_timestamp_peer = peer.clone();
542            tampered_same_timestamp_peer.pow_initial.salt = crate::tools::types::Salt::zero();
543            assert!(peer_tracker.add_peer(tampered_same_timestamp_peer).is_ok());
544            assert!(!peer_tracker.needs_flush(), "a non-newer copy must not trigger a flush");
545            assert_eq!(peer, peer_tracker.peers()[0]);
546
547            // A newer but invalid copy does reach verify and is rejected.
548            let mut tampered_newer_peer = peer.clone();
549            tampered_newer_peer.timestamp = TimeMillis(peer.timestamp.0 + 1);
550            assert!(peer_tracker.add_peer(tampered_newer_peer).is_err());
551            assert_eq!(peer, peer_tracker.peers()[0]);
552            Ok(())
553        }
554
555        #[tokio::test]
556        async fn load_drops_far_future_peers_and_trims_to_max_peers() -> anyhow::Result<()> {
557            let (runtime_services, _) = manual_runtime_services();
558            let client_storage = MemClientStorage::new().await?;
559            const MAX_PEERS: usize = 3;
560
561            let mut stored_peers = Vec::new();
562            for _ in 0..MAX_PEERS + 2 {
563                let (_, peer) = make_peer(&runtime_services).await?;
564                stored_peers.push(peer);
565            }
566            let (far_future_server_id, mut far_future_peer) = make_peer(&runtime_services).await?;
567            let future_time_provider = ManualTimeProvider::new(START_TIME_MILLIS + config::PEER_MAX_FUTURE_TIMESTAMP_MILLIS.const_mul(2));
568            far_future_peer.sign(&future_time_provider, &far_future_server_id.keys.signature_key)?;
569            stored_peers.push(far_future_peer.clone());
570
571            let now = runtime_services.time_provider.current_time_millis();
572            client_storage::put_struct(client_storage.as_ref(), BUCKET_PEER, "peers", &stored_peers, now).await?;
573
574            let peer_tracker = PeerTracker::new_with_max_peers(runtime_services.clone(), client_storage, MAX_PEERS).await?;
575            assert_eq!(MAX_PEERS, peer_tracker.len());
576            assert!(!peer_ids(&peer_tracker).contains(&far_future_peer.id), "far-future stored peer must be dropped");
577            assert!(peer_tracker.needs_flush(), "trimmed set must be persisted");
578            for window in peer_tracker.peers().windows(2) {
579                assert!(window[0].id < window[1].id, "loaded peers must be sorted by id");
580            }
581            Ok(())
582        }
583
584        #[tokio::test]
585        async fn add_peers_considers_at_most_the_per_response_cap() -> anyhow::Result<()> {
586            let runtime_services = RuntimeServices::default_for_testing();
587            let client_storage = MemClientStorage::new().await?;
588            let peer_tracker: SharedPeerTracker = Arc::new(parking_lot::RwLock::new(PeerTracker::new(runtime_services.clone(), client_storage).await?));
589
590            let mut response_peers = Vec::new();
591            for _ in 0..config::CLIENT_MAX_PEERS_CONSIDERED_PER_RESPONSE + 3 {
592                let (_, peer) = make_peer(&runtime_services).await?;
593                response_peers.push(peer);
594            }
595
596            let mut peer_iterator = PeerIterator::new(peer_tracker.clone(), Id::random(), 1, None);
597            peer_iterator.add_peers(response_peers);
598            assert_eq!(config::CLIENT_MAX_PEERS_CONSIDERED_PER_RESPONSE, peer_tracker.read().len());
599            Ok(())
600        }
601
602        #[tokio::test]
603        async fn insert_verified_rechecks_timestamp() -> anyhow::Result<()> {
604            let (runtime_services, manual_time_provider) = manual_runtime_services();
605            let client_storage = MemClientStorage::new().await?;
606            let mut peer_tracker = PeerTracker::new(runtime_services.clone(), client_storage).await?;
607
608            let (server_id, older_peer) = make_peer(&runtime_services).await?;
609            manual_time_provider.set_time(START_TIME_MILLIS + MILLIS_IN_DAY);
610            let mut newer_peer = older_peer.clone();
611            newer_peer.sign(runtime_services.time_provider.as_ref(), &server_id.keys.signature_key)?;
612            assert!(newer_peer.timestamp > older_peer.timestamp, "test setup: the re-signed copy must be newer");
613
614            // Another walk verified the older copy, but the newer one was inserted first: the older one must not overwrite it
615            peer_tracker.insert_verified(newer_peer.clone());
616            peer_tracker.insert_verified(older_peer);
617            assert_eq!(vec![newer_peer], *peer_tracker.peers());
618            Ok(())
619        }
620
621        #[tokio::test]
622        async fn needs_verification_skips_known_peers_that_are_not_newer() -> anyhow::Result<()> {
623            let (runtime_services, manual_time_provider) = manual_runtime_services();
624            let client_storage = MemClientStorage::new().await?;
625            let mut peer_tracker = PeerTracker::new(runtime_services.clone(), client_storage).await?;
626
627            let (server_id, peer) = make_peer(&runtime_services).await?;
628            assert!(peer_tracker.needs_verification(&peer), "an unknown peer needs verifying");
629            peer_tracker.add_peer(peer.clone())?;
630            assert!(!peer_tracker.needs_verification(&peer), "the same copy does not");
631
632            manual_time_provider.set_time(START_TIME_MILLIS + MILLIS_IN_DAY);
633            let mut newer_peer = peer.clone();
634            newer_peer.sign(runtime_services.time_provider.as_ref(), &server_id.keys.signature_key)?;
635            assert!(peer_tracker.needs_verification(&newer_peer), "a newer copy does");
636            Ok(())
637        }
638
639        #[tokio::test]
640        async fn flush_snapshot_clears_and_mark_restores_flag() -> anyhow::Result<()> {
641            let runtime_services = RuntimeServices::default_for_testing();
642            let client_storage = MemClientStorage::new().await?;
643            let mut peer_tracker = PeerTracker::new(runtime_services.clone(), client_storage).await?;
644            assert!(peer_tracker.take_flush_snapshot().is_none(), "nothing to flush yet");
645
646            let (_, peer) = make_peer(&runtime_services).await?;
647            peer_tracker.add_peer(peer.clone())?;
648            assert_eq!(Some(vec![peer]), peer_tracker.take_flush_snapshot());
649            assert!(!peer_tracker.needs_flush());
650            assert!(peer_tracker.take_flush_snapshot().is_none());
651
652            peer_tracker.mark_needs_flush();
653            assert!(peer_tracker.needs_flush());
654            Ok(())
655        }
656
657        #[tokio::test]
658        async fn shared_flush_persists_peers() -> anyhow::Result<()> {
659            let runtime_services = RuntimeServices::default_for_testing();
660            let client_storage = MemClientStorage::new().await?;
661            let peer_tracker: SharedPeerTracker = Arc::new(parking_lot::RwLock::new(PeerTracker::new(runtime_services.clone(), client_storage.clone()).await?));
662
663            for _ in 0..3 {
664                let (_, peer) = make_peer(&runtime_services).await?;
665                peer_tracker.write().add_peer(peer)?;
666            }
667            PeerTracker::flush(&peer_tracker).await?;
668            assert!(!peer_tracker.read().needs_flush());
669
670            let reloaded_peer_tracker = PeerTracker::new(runtime_services.clone(), client_storage).await?;
671            assert_eq!(*peer_tracker.read().peers(), *reloaded_peer_tracker.peers());
672            Ok(())
673        }
674    }
675
676    /// peers() must stay sorted by id after arbitrary insertion order, because
677    /// add_peer uses binary_search_by_key to dedupe.
678    #[tokio::test]
679    async fn add_peer_keeps_peers_sorted_by_id() -> anyhow::Result<()> {
680        let runtime_services = RuntimeServices::default_for_testing();
681        let client_storage = MemClientStorage::new().await?;
682        let mut peer_tracker = PeerTracker::new(runtime_services.clone(), client_storage.clone()).await?;
683
684        const NUM_PEERS: usize = 30;
685        for _ in 0..NUM_PEERS {
686            let server_id = ServerId::new("own_pow", runtime_services.time_provider.as_ref(), config::SERVER_KEY_POW_MIN, true, runtime_services.pow_generator.as_ref()).await?;
687            let peer = server_id.to_peer(runtime_services.time_provider.as_ref())?;
688            peer_tracker.add_peer(peer)?;
689        }
690
691        let peers = peer_tracker.peers();
692        assert_eq!(NUM_PEERS, peers.len());
693        for window in peers.windows(2) {
694            assert!(window[0].id < window[1].id, "peers must be sorted by id; got {} then {}", window[0].id, window[1].id);
695        }
696
697        Ok(())
698    }
699}
hashiverse_lib/client/peer_tracker/peer_tracker.rs

hashiverse_lib/client/peer_tracker/
peer_tracker.rs
