hashiverse_lib/client/peer_tracker/
peer_iterator.rs

1//! # Stateful peer walk by XOR distance
2//!
3//! `PeerIterator` is the execution primitive for every DHT operation: given a target
4//! [`crate::tools::types::Id`] and a
5//! [`crate::client::peer_tracker::peer_tracker::PeerTracker`], hand back peers in order
6//! of decreasing closeness while remembering which ones have already been tried.
7//!
8//! Two knobs tune the walk:
9//!
10//! - **High-watermark** on [`crate::tools::tools::LeadingAgreementBits`] — once `N`
11//!   iterations have passed without finding a peer closer than the best so far, the
12//!   iterator gives up. This is the standard Kademlia "no further progress possible"
13//!   signal.
14//! - **Cache radius** — supplied by
15//!   [`crate::client::caching::cache_radius_tracker`]. Peers closer than the recorded
16//!   cache radius are skipped, because whatever we'd fetch from them is already cached
17//!   further out; fetching again just hammers the closest nodes.
18
19use crate::client::peer_tracker::peer_tracker::{PeerTracker, SharedPeerTracker};
20use crate::protocol::peer::Peer;
21use crate::tools::{config, tools};
22use crate::tools::tools::LeadingAgreementBits;
23use crate::tools::types::Id;
24use log::warn;
25use std::collections::HashSet;
26
27/// Holds only a [`SharedPeerTracker`] handle: each call locks the tracker briefly and synchronously, so concurrent walks
28/// interleave freely and nothing is locked while the caller awaits its RPCs.
29pub struct PeerIterator {
30    tracker: SharedPeerTracker,
31    bucket_location_id: Id,
32    max_iterations_since_high_watermark: usize,
33    peers_already_queried: HashSet<Id>,
34    high_watermark: LeadingAgreementBits,
35    iterations_since_high_watermark: usize,
36    cache_radius: Option<LeadingAgreementBits>,
37}
38
39impl PeerIterator {
40    pub fn new(tracker: SharedPeerTracker, bucket_location_id: Id, max_iterations_since_high_watermark: usize, cache_radius: Option<LeadingAgreementBits>) -> Self {
41        Self {
42            tracker,
43            bucket_location_id,
44            max_iterations_since_high_watermark,
45            peers_already_queried: HashSet::new(),
46            high_watermark: 0,
47            iterations_since_high_watermark: 0,
48            cache_radius,
49        }
50    }
51    pub fn next_peer(&mut self) -> Option<(Peer, LeadingAgreementBits)> {
52        loop {
53            let nearest_peer = self
54                .tracker
55                .read()
56                .peers()
57                .iter()
58                .filter(|peer| !self.peers_already_queried.contains(&peer.id))
59                .map(|peer| (peer, tools::leading_agreement_bits_xor(&self.bucket_location_id.0, &peer.id.0)))
60                .filter(|(_, lab)| self.cache_radius.is_none_or(|r| *lab < r))
61                .max_by_key(|peer| peer.1)
62                .map(|(peer, lab)| (peer.clone(), lab));
63
64            match nearest_peer {
65                Some(nearest_peer) => {
66                    self.peers_already_queried.insert(nearest_peer.0.id);
67
68                    if nearest_peer.1 > self.high_watermark {
69                        self.high_watermark = nearest_peer.1;
70                        self.iterations_since_high_watermark = 0;
71                    }
72                    else {
73                        self.iterations_since_high_watermark += 1;
74                        if self.iterations_since_high_watermark > self.max_iterations_since_high_watermark {
75                            return None;
76                        }
77                    }
78
79                    // Each successful return opens one more ring of closer peers on the next call.
80                    if let Some(r) = &mut self.cache_radius {
81                        *r = (*r + 1).min(256);
82                    }
83
84                    return Some(nearest_peer);
85                }
86                None => {
87                    // No unvisited peer passes the current radius filter.
88                    // If there are no unvisited peers at all, we are done.
89                    let any_unvisited = self.tracker.read().peers().iter().any(|p| !self.peers_already_queried.contains(&p.id));
90                    if !any_unvisited {
91                        return None;
92                    }
93                    // Allow the next ring of closer peers and retry.
94                    //
95                    // NOTE: must take `&mut` via `as_mut()`, NOT `&mut self.cache_radius?`.
96                    // The latter applies `?` first, which copies the inner `i32` out (because
97                    // `LeadingAgreementBits: Copy`), so `&mut` then borrows a stack temporary
98                    // and the mutation never reaches `self.cache_radius`. That regression
99                    // (introduced in commit c1e734e) caused an infinite loop whenever the
100                    // initial cache_radius was 0.
101                    let r = self.cache_radius.as_mut()?;
102                    *r = (*r + 1).min(256);
103                }
104            }
105        }
106    }
107
108    pub fn iterations_since_high_watermark(&self) -> usize {
109        self.iterations_since_high_watermark
110    }
111
112    pub fn add_peers(&mut self, peers: Vec<Peer>) {
113        if peers.len() > config::CLIENT_MAX_PEERS_CONSIDERED_PER_RESPONSE {
114            warn!("ignoring {} of {} suggested peers (cap is {})", peers.len() - config::CLIENT_MAX_PEERS_CONSIDERED_PER_RESPONSE, peers.len(), config::CLIENT_MAX_PEERS_CONSIDERED_PER_RESPONSE);
115        }
116
117        // Dedupe under a read lock, verify (the expensive part) with no lock held, then insert under one short write lock
118        let (candidate_peers, current_time_millis) = {
119            let tracker = self.tracker.read();
120            let candidate_peers: Vec<Peer> = peers.into_iter().take(config::CLIENT_MAX_PEERS_CONSIDERED_PER_RESPONSE).filter(|peer| tracker.needs_verification(peer)).collect();
121            (candidate_peers, tracker.current_time_millis())
122        };
123
124        let verified_peers: Vec<Peer> = candidate_peers
125            .into_iter()
126            .filter(|peer| match PeerTracker::verify_candidate(peer, current_time_millis) {
127                Ok(()) => true,
128                Err(e) => {
129                    warn!("not adding invalid peer: {}", e);
130                    false
131                }
132            })
133            .collect();
134
135        if !verified_peers.is_empty() {
136            let mut tracker = self.tracker.write();
137            for peer in verified_peers {
138                tracker.insert_verified(peer);
139            }
140        }
141    }
142    pub fn remove_peer(&mut self, peer: &Peer) {
143        self.tracker.write().remove_peer(peer);
144    }
145}
146
147pub struct ConvergeToLocationVisitResult {
148    pub done: bool,                  // Stop iterating
149    pub peer_unavailable: bool,      // Indicate that this peer has a problem of some sort and should be removed
150    pub peers_discovered: Vec<Peer>, // Supply some newly discovered Peers to add to the iterations
151}
152
153#[async_trait::async_trait]
154pub trait ConvergeToLocationVisitor: Send + Sync {
155    async fn on_peer(&mut self, peer: &Peer) -> anyhow::Result<ConvergeToLocationVisitResult>;
156}
157
158#[cfg(test)]
159mod tests {
160    use crate::client::client_storage::mem_client_storage::MemClientStorage;
161    use crate::client::peer_tracker::peer_tracker::{PeerTracker, SharedPeerTracker};
162    use std::sync::Arc;
163    use crate::tools::buckets::{BUCKET_DURATIONS, BucketLocation, BucketType, generate_bucket_location};
164    use crate::tools::config;
165    use crate::tools::runtime_services::RuntimeServices;
166    use crate::tools::server_id::ServerId;
167    use crate::tools::time::{DurationMillis, TimeMillis};
168    use crate::tools::types::Id;
169
170    #[tokio::test]
171    async fn converge_basics_test() -> anyhow::Result<()> {
172        let runtime_services = RuntimeServices::default_for_testing();
173        let client_storage = MemClientStorage::new().await?;
174        let peer_tracker: SharedPeerTracker = Arc::new(parking_lot::RwLock::new(PeerTracker::new(runtime_services.clone(), client_storage.clone()).await?));
175
176        const NUM_PEERS: usize = 100;
177
178        {
179            for _ in 0..NUM_PEERS {
180                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?;
181                let peer = server_id.to_peer(runtime_services.time_provider.as_ref())?;
182                peer_tracker.write().add_peer(peer)?;
183            }
184            assert_eq!(NUM_PEERS, peer_tracker.read().len());
185        }
186
187        // Now iterate through them all
188        {
189            let bucket_location = generate_bucket_location(BucketType::User, Id::random(), BUCKET_DURATIONS[0], runtime_services.time_provider.current_time_millis())?;
190            let mut count = 0;
191            let mut peer_iter = PeerTracker::iterate_to_location(&peer_tracker, bucket_location.location_id, usize::MAX, None).await?;
192            while let Some(_peer) = peer_iter.next_peer() {
193                count += 1;
194            }
195            assert_eq!(NUM_PEERS, count);
196        };
197
198        // Now iterate through them, but bail after the first
199        {
200            let bucket_location = generate_bucket_location(BucketType::User, Id::random(), BUCKET_DURATIONS[0], runtime_services.time_provider.current_time_millis())?;
201            let mut count = 0;
202            let mut peer_iter = PeerTracker::iterate_to_location(&peer_tracker, bucket_location.location_id, usize::MAX, None).await?;
203            if peer_iter.next_peer().is_some() {
204                count += 1;
205            }
206            assert_eq!(1, count);
207        };
208
209        // Now iterate through them, but delete half of them
210        {
211            let bucket_location = generate_bucket_location(BucketType::User, Id::random(), BUCKET_DURATIONS[0], runtime_services.time_provider.current_time_millis())?;
212            let mut count = 0;
213            let mut peer_iter = PeerTracker::iterate_to_location(&peer_tracker, bucket_location.location_id, usize::MAX, None).await?;
214            while let Some((peer, _)) = peer_iter.next_peer() {
215                count += 1;
216                if 0 == count % 2 {
217                    peer_iter.remove_peer(&peer);
218                }
219            }
220            assert_eq!(NUM_PEERS, count);
221            assert_eq!(NUM_PEERS / 2, peer_tracker.read().len());
222        }
223
224        Ok(())
225    }
226
227    #[tokio::test]
228    async fn converge_termination_test() -> anyhow::Result<()> {
229        let runtime_services = RuntimeServices::default_for_testing();
230        let client_storage = MemClientStorage::new().await?;
231        let peer_tracker: SharedPeerTracker = Arc::new(parking_lot::RwLock::new(PeerTracker::new(runtime_services.clone(), client_storage.clone()).await?));
232
233        const NUM_PEERS: usize = 100;
234
235        {
236            for _ in 0..NUM_PEERS {
237                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?;
238                let peer = server_id.to_peer(runtime_services.time_provider.as_ref())?;
239                peer_tracker.write().add_peer(peer)?;
240            }
241            assert_eq!(NUM_PEERS, peer_tracker.read().len());
242        }
243
244        {
245            let bucket_location = generate_bucket_location(BucketType::User, Id::random(), BUCKET_DURATIONS[0], runtime_services.time_provider.current_time_millis())?;
246            let mut count = 0;
247            let mut peer_iter = PeerTracker::iterate_to_location(&peer_tracker, bucket_location.location_id, 3, None).await?;
248            while let Some(_peer) = peer_iter.next_peer() {
249                count += 1;
250            }
251            assert_eq!(3 + 1, count);
252        }
253
254        Ok(())
255    }
256
257    #[tokio::test]
258    async fn converge_insertions_test() -> anyhow::Result<()> {
259        let runtime_services = RuntimeServices::default_for_testing();
260        let client_storage = MemClientStorage::new().await?;
261        let peer_tracker: SharedPeerTracker = Arc::new(parking_lot::RwLock::new(PeerTracker::new(runtime_services.clone(), client_storage.clone()).await?));
262
263        const NUM_PEERS: usize = 100;
264
265        {
266            for _ in 0..NUM_PEERS {
267                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?;
268                let peer = server_id.to_peer(runtime_services.time_provider.as_ref())?;
269                peer_tracker.write().add_peer(peer)?;
270            }
271            assert_eq!(NUM_PEERS, peer_tracker.read().len());
272        }
273
274        // Now iterate through them, but add a few more peers
275        {
276            let bucket_location = generate_bucket_location(BucketType::User, Id::random(), BUCKET_DURATIONS[0], runtime_services.time_provider.current_time_millis())?;
277            let mut count = 0;
278            let mut peer_iter = PeerTracker::iterate_to_location(&peer_tracker, bucket_location.location_id, usize::MAX, None).await?;
279            while let Some(_peer) = peer_iter.next_peer() {
280                count += 1;
281
282                if 0 == count % 10 {
283                    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?;
284                    let peer = server_id.to_peer(runtime_services.time_provider.as_ref())?;
285                    peer_iter.add_peers(vec![peer]);
286                }
287
288                if 50 == count {
289                    break;
290                }
291            }
292
293            assert_eq!(50, count);
294            assert_eq!(NUM_PEERS + 5, peer_tracker.read().len());
295        }
296
297        Ok(())
298    }
299
300    #[tokio::test]
301    async fn converge_targeting_test() -> anyhow::Result<()> {
302        let runtime_services = RuntimeServices::default_for_testing();
303        let client_storage = MemClientStorage::new().await?;
304        let peer_tracker: SharedPeerTracker = Arc::new(parking_lot::RwLock::new(PeerTracker::new(runtime_services.clone(), client_storage.clone()).await?));
305
306        const NUM_PEERS: usize = 100;
307
308        {
309            for _ in 0..NUM_PEERS {
310                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?;
311                let peer = server_id.to_peer(runtime_services.time_provider.as_ref())?;
312                peer_tracker.write().add_peer(peer)?;
313            }
314            assert_eq!(NUM_PEERS, peer_tracker.read().len());
315        }
316
317        // This is the peer we are actually targeting
318        let target_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?;
319        let target_peer = target_server_id.to_peer(runtime_services.time_provider.as_ref())?;
320
321        {
322            const PEER_DISCOVERY_I: usize = 37usize;
323            const PEER_DISCOVERY_I_PLUS_1: usize = PEER_DISCOVERY_I + 1;
324
325            let bucket_location = {
326                let mut location_id = target_peer.id;
327                for i in 10..31 {
328                    location_id.0[i] = 0u8;
329                }
330                BucketLocation {
331                    bucket_type: BucketType::User,
332                    base_id: location_id,
333                    duration: DurationMillis::zero(),
334                    bucket_time_millis: TimeMillis::zero(),
335                    location_id,
336                }
337            };
338
339            let mut count = 0;
340            let mut peer_iter = PeerTracker::iterate_to_location(&peer_tracker, bucket_location.location_id, usize::MAX, None).await?;
341            while let Some((peer, _)) = peer_iter.next_peer() {
342                count += 1;
343                match count {
344                    PEER_DISCOVERY_I => {
345                        peer_iter.add_peers(vec![target_peer.clone()]);
346                    }
347                    PEER_DISCOVERY_I_PLUS_1 => {
348                        if peer.id != target_peer.id {
349                            anyhow::bail!("peer is not the one we expected");
350                        }
351                        break;
352                    }
353                    _ => {}
354                }
355            }
356
357            assert_eq!(PEER_DISCOVERY_I_PLUS_1, count);
358            assert_eq!(NUM_PEERS + 1, peer_tracker.read().len());
359        }
360
361        Ok(())
362    }
363
364    /// Verify that `cache_radius` starts by skipping peers inside the radius, then opens up
365    /// one ring per step so that closer peers are eventually visited too.
366    #[tokio::test]
367    async fn converge_cache_radius_test() -> anyhow::Result<()> {
368        let runtime_services = RuntimeServices::default_for_testing();
369        let client_storage = MemClientStorage::new().await?;
370        let peer_tracker: SharedPeerTracker = Arc::new(parking_lot::RwLock::new(PeerTracker::new(runtime_services.clone(), client_storage.clone()).await?));
371
372        let location_id = Id::zero();
373
374        const NUM_PEERS: usize = 100;
375        let mut labs_added: Vec<crate::tools::tools::LeadingAgreementBits> = Vec::new();
376        for _ in 0..NUM_PEERS {
377            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?;
378            let peer = server_id.to_peer(runtime_services.time_provider.as_ref())?;
379            let lab = crate::tools::tools::leading_agreement_bits_xor(&location_id.0, &peer.id.0);
380            labs_added.push(lab);
381            peer_tracker.write().add_peer(peer)?;
382        }
383        assert_eq!(NUM_PEERS, peer_tracker.read().len());
384
385        let mut sorted_labs = labs_added.clone();
386        sorted_labs.sort();
387        let cache_radius = sorted_labs[NUM_PEERS / 2];
388
389        let mut labs_visited: Vec<crate::tools::tools::LeadingAgreementBits> = Vec::new();
390        let mut peer_iter = PeerTracker::iterate_to_location(&peer_tracker, location_id, usize::MAX, Some(cache_radius)).await?;
391        while let Some((_, lab)) = peer_iter.next_peer() {
392            labs_visited.push(lab);
393        }
394
395        assert_eq!(NUM_PEERS, labs_visited.len(), "all peers should be visited");
396
397        let has_outside_peers = labs_added.iter().any(|&lab| lab < cache_radius);
398        if has_outside_peers {
399            assert!(labs_visited[0] < cache_radius, "first peer should be outside the initial cache zone, got lab={} cache_radius={}", labs_visited[0], cache_radius);
400        }
401
402        Ok(())
403    }
404
405    /// Two walks over one shared tracker interleave step by step; each sees the other's additions and removals.
406    #[tokio::test]
407    async fn two_iterators_interleave() -> anyhow::Result<()> {
408        let runtime_services = RuntimeServices::default_for_testing();
409        let client_storage = MemClientStorage::new().await?;
410        let peer_tracker: SharedPeerTracker = Arc::new(parking_lot::RwLock::new(PeerTracker::new(runtime_services.clone(), client_storage.clone()).await?));
411
412        const NUM_PEERS: usize = 20;
413        for _ in 0..NUM_PEERS {
414            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?;
415            peer_tracker.write().add_peer(server_id.to_peer(runtime_services.time_provider.as_ref())?)?;
416        }
417
418        let mut peer_iter_a = PeerTracker::iterate_to_location(&peer_tracker, Id::random(), usize::MAX, None).await?;
419        let mut peer_iter_b = PeerTracker::iterate_to_location(&peer_tracker, Id::random(), usize::MAX, None).await?;
420
421        // Walk A removes the peers it visits; walk B adds a fresh peer each step
422        let mut visited_by_a = 0;
423        let mut added_by_b = 0;
424        let mut a_done = false;
425        let mut b_done = false;
426        while !(a_done && b_done) {
427            if !a_done {
428                match peer_iter_a.next_peer() {
429                    Some((peer, _)) => {
430                        visited_by_a += 1;
431                        peer_iter_a.remove_peer(&peer);
432                    }
433                    None => a_done = true,
434                }
435            }
436            if !b_done {
437                match peer_iter_b.next_peer() {
438                    Some(_) if added_by_b < 5 => {
439                        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?;
440                        peer_iter_b.add_peers(vec![server_id.to_peer(runtime_services.time_provider.as_ref())?]);
441                        added_by_b += 1;
442                    }
443                    Some(_) => {}
444                    None => b_done = true,
445                }
446            }
447        }
448
449        // A visited (and removed) every peer that existed while it walked, including those B added before A finished
450        assert!(visited_by_a >= NUM_PEERS, "A must visit at least the initial peers, visited {}", visited_by_a);
451        assert_eq!(NUM_PEERS + added_by_b - visited_by_a, peer_tracker.read().len());
452        Ok(())
453    }
454
455    /// An outstanding iterator holds no lock, so the tracker stays readable and writable between steps.
456    #[tokio::test]
457    async fn tracker_unlocked_while_iterator_alive() -> anyhow::Result<()> {
458        let runtime_services = RuntimeServices::default_for_testing();
459        let client_storage = MemClientStorage::new().await?;
460        let peer_tracker: SharedPeerTracker = Arc::new(parking_lot::RwLock::new(PeerTracker::new(runtime_services.clone(), client_storage.clone()).await?));
461        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?;
462        peer_tracker.write().add_peer(server_id.to_peer(runtime_services.time_provider.as_ref())?)?;
463
464        let mut peer_iter = PeerTracker::iterate_to_location(&peer_tracker, Id::random(), usize::MAX, None).await?;
465        assert!(peer_tracker.try_write().is_some(), "no lock may be held after construction");
466        assert!(peer_iter.next_peer().is_some());
467        assert!(peer_tracker.try_write().is_some(), "no lock may be held between steps");
468        Ok(())
469    }
470}
hashiverse_lib/client/peer_tracker/peer_iterator.rs

hashiverse_lib/client/peer_tracker/
peer_iterator.rs
