tor_guardmgr/fallback/
set.rs

Help
1//! Declare the [`FallbackState`] type, which is used to store a set of FallbackDir.
2
3use crate::skew::SkewObservation;
4use rand::seq::IteratorRandom;
5use std::time::{Duration, Instant};
6use tor_linkspec::HasRelayIds;
7
8use super::{DirStatus, FallbackDir, FallbackDirBuilder};
9use crate::fallback::default_fallbacks;
10use crate::{ids::FallbackId, PickGuardError};
11use tor_basic_utils::iter::{FilterCount, IteratorExt as _};
12use tor_config::define_list_builder_helper;
13
14/// A list of fallback directories.
15///
16/// Fallback directories (represented by [`FallbackDir`]) are used by Tor
17/// clients when they don't already have enough other directory information to
18/// contact the network.
19#[derive(Debug, Clone, Default, PartialEq, Eq)]
20pub struct FallbackList {
21    /// The underlying fallbacks in this set.
22    fallbacks: Vec<FallbackDir>,
23}
24
25impl<T: IntoIterator<Item = FallbackDir>> From<T> for FallbackList {
26    fn from(fallbacks: T) -> Self {
27        FallbackList {
28            fallbacks: fallbacks.into_iter().collect(),
29        }
30    }
31}
32
33define_list_builder_helper! {
34    // pub because tor-dirmgr needs it for NetworkConfig.fallback_caches
35    pub struct FallbackListBuilder {
36        pub(crate) fallbacks: [FallbackDirBuilder],
37    }
38    built: FallbackList = FallbackList { fallbacks };
39    default = default_fallbacks();
40}
41
42impl FallbackList {
43    /// Return the number of fallbacks in this list.
44    pub fn len(&self) -> usize {
45        self.fallbacks.len()
46    }
47    /// Return true if there are no fallbacks in this list.
48    pub fn is_empty(&self) -> bool {
49        self.fallbacks.is_empty()
50    }
51    /// Return a random member of this list.
52    pub fn choose<R: rand::Rng>(&self, rng: &mut R) -> Result<&FallbackDir, PickGuardError> {
53        self.fallbacks
54            .iter()
55            .choose(rng)
56            .ok_or(PickGuardError::NoCandidatesAvailable)
57    }
58}
59
60/// A set of fallback directories, in usable form.
61#[derive(Debug, Clone)]
62pub(crate) struct FallbackState {
63    /// The list of fallbacks in the set.
64    ///
65    /// We require that these are sorted and unique by (ED,RSA) keys.
66    fallbacks: Vec<Entry>,
67}
68
69/// Wrapper type for FallbackDir converted into crate::Guard, and the status
70/// information that we store about it.
71///
72/// Defines a sort order to ensure that we can look up fallback directories by
73/// binary search on keys.
74#[derive(Debug, Clone)]
75pub(super) struct Entry {
76    /// The inner fallback directory.
77    fallback: FallbackDir,
78
79    /// Whether the directory is currently usable, and if not, when we can retry
80    /// it.
81    status: DirStatus,
82    /// The latest clock skew observation we have from this fallback directory
83    /// (if any).
84    clock_skew: Option<SkewObservation>,
85}
86
87/// Least amount of time we'll wait before retrying a fallback cache.
88//
89// TODO: we may want to make this configurable to a smaller value for chutney networks.
90const FALLBACK_RETRY_FLOOR: Duration = Duration::from_secs(150);
91
92impl From<FallbackDir> for Entry {
93    fn from(fallback: FallbackDir) -> Self {
94        let status = DirStatus::new(FALLBACK_RETRY_FLOOR);
95        Entry {
96            fallback,
97            status,
98            clock_skew: None,
99        }
100    }
101}
102
103impl HasRelayIds for Entry {
104    fn identity(
105        &self,
106        key_type: tor_linkspec::RelayIdType,
107    ) -> Option<tor_linkspec::RelayIdRef<'_>> {
108        self.fallback.identity(key_type)
109    }
110}
111
112impl From<&FallbackList> for FallbackState {
113    fn from(list: &FallbackList) -> Self {
114        let mut fallbacks: Vec<Entry> = list.fallbacks.iter().map(|fb| fb.clone().into()).collect();
115        fallbacks.sort_by(|x, y| x.cmp_by_relay_ids(y));
116        fallbacks.dedup_by(|x, y| x.same_relay_ids(y));
117        FallbackState { fallbacks }
118    }
119}
120
121impl FallbackState {
122    /// Return a random member of this FallbackSet that's usable at `now`.
123    pub(crate) fn choose<R: rand::Rng>(
124        &self,
125        rng: &mut R,
126        now: Instant,
127        filter: &crate::GuardFilter,
128    ) -> Result<&FallbackDir, PickGuardError> {
129        if self.fallbacks.is_empty() {
130            return Err(PickGuardError::NoCandidatesAvailable);
131        }
132
133        let mut running = FilterCount::default();
134        let mut filtered = FilterCount::default();
135
136        self.fallbacks
137            .iter()
138            .filter_cnt(&mut running, |ent| ent.status.usable_at(now))
139            .filter_cnt(&mut filtered, |ent| filter.permits(&ent.fallback))
140            .choose(rng)
141            .map(|ent| &ent.fallback)
142            .ok_or_else(|| PickGuardError::AllFallbacksDown {
143                retry_at: self.next_retry(),
144                running,
145                filtered,
146            })
147    }
148
149    /// Return the next time at which any member of this set will become ready.
150    ///
151    /// Returns None if no elements are failing.
152    fn next_retry(&self) -> Option<Instant> {
153        self.fallbacks
154            .iter()
155            .filter_map(|ent| ent.status.next_retriable())
156            .min()
157    }
158
159    /// Return a reference to the entry whose identity is `id`, if there is one.
160    fn get(&self, id: &FallbackId) -> Option<&Entry> {
161        match self.fallbacks.binary_search_by(|e| e.cmp_by_relay_ids(id)) {
162            Ok(idx) => Some(&self.fallbacks[idx]),
163            Err(_) => None,
164        }
165    }
166
167    /// Return a mutable reference to the entry whose identity is `id`, if there is one.
168    fn get_mut(&mut self, id: &FallbackId) -> Option<&mut Entry> {
169        match self.fallbacks.binary_search_by(|e| e.cmp_by_relay_ids(id)) {
170            Ok(idx) => Some(&mut self.fallbacks[idx]),
171            Err(_) => None,
172        }
173    }
174
175    /// Return true if this set contains some entry with the given `id`.
176    pub(crate) fn contains(&self, id: &FallbackId) -> bool {
177        self.get(id).is_some()
178    }
179
180    /// Record that a success has occurred for the fallback with the given
181    /// identity.
182    ///
183    /// Be aware that for fallbacks, we only count a successful directory
184    /// operation as a success: a circuit success is not enough.
185    pub(crate) fn note_success(&mut self, id: &FallbackId) {
186        if let Some(entry) = self.get_mut(id) {
187            entry.status.note_success();
188        }
189    }
190
191    /// Record that a failure has occurred for the fallback with the given
192    /// identity.
193    pub(crate) fn note_failure(&mut self, id: &FallbackId, now: Instant) {
194        if let Some(entry) = self.get_mut(id) {
195            entry.status.note_failure(now);
196        }
197    }
198
199    /// Consume `other` and copy all of its fallback status entries into the corresponding entries for `self`.
200    pub(crate) fn take_status_from(&mut self, other: FallbackState) {
201        use itertools::EitherOrBoth::Both;
202
203        itertools::merge_join_by(self.fallbacks.iter_mut(), other.fallbacks, |a, b| {
204            a.fallback.cmp_by_relay_ids(&b.fallback)
205        })
206        .for_each(|entry| {
207            if let Both(entry, other) = entry {
208                debug_assert!(entry.fallback.same_relay_ids(&other.fallback));
209                entry.status = other.status;
210            }
211        });
212    }
213
214    /// Record that a given fallback has told us about clock skew.
215    pub(crate) fn note_skew(&mut self, id: &FallbackId, observation: SkewObservation) {
216        if let Some(entry) = self.get_mut(id) {
217            entry.clock_skew = Some(observation);
218        }
219    }
220
221    /// Return an iterator over all the clock skew observations we've made for fallback directories
222    pub(crate) fn skew_observations(&self) -> impl Iterator<Item = &SkewObservation> {
223        self.fallbacks
224            .iter()
225            .filter_map(|fb| fb.clock_skew.as_ref())
226    }
227}
228
229#[cfg(test)]
230mod test {
231    // @@ begin test lint list maintained by maint/add_warning @@
232    #![allow(clippy::bool_assert_comparison)]
233    #![allow(clippy::clone_on_copy)]
234    #![allow(clippy::dbg_macro)]
235    #![allow(clippy::mixed_attributes_style)]
236    #![allow(clippy::print_stderr)]
237    #![allow(clippy::print_stdout)]
238    #![allow(clippy::single_char_pattern)]
239    #![allow(clippy::unwrap_used)]
240    #![allow(clippy::unchecked_duration_subtraction)]
241    #![allow(clippy::useless_vec)]
242    #![allow(clippy::needless_pass_by_value)]
243    //! <!-- @@ end test lint list maintained by maint/add_warning @@ -->
244
245    use super::*;
246    use rand::Rng;
247    use tor_basic_utils::test_rng::testing_rng;
248
249    /// Construct a `FallbackDir` with random identity keys and addresses.
250    ///
251    /// Since there are 416 bits of random id here, the risk of collision is
252    /// negligible.
253    fn rand_fb<R: Rng>(rng: &mut R) -> FallbackDir {
254        let ed: [u8; 32] = rng.random();
255        let rsa: [u8; 20] = rng.random();
256        let ip: u32 = rng.random();
257        let mut bld = FallbackDir::builder();
258        bld.ed_identity(ed.into())
259            .rsa_identity(rsa.into())
260            .orports()
261            .push(std::net::SocketAddrV4::new(ip.into(), 9090).into());
262        bld.build().unwrap()
263    }
264
265    #[test]
266    fn construct_fallback_set() {
267        use rand::seq::SliceRandom;
268        use std::cmp::Ordering as O;
269
270        // fabricate some fallbacks.
271        let mut rng = testing_rng();
272        let fbs = vec![
273            rand_fb(&mut rng),
274            rand_fb(&mut rng),
275            rand_fb(&mut rng),
276            rand_fb(&mut rng),
277        ];
278        let fb_other = rand_fb(&mut rng);
279        let id_other = FallbackId::from_relay_ids(&fb_other);
280
281        // basic case: construct a set
282        let list: FallbackList = fbs.clone().into();
283        assert!(!list.is_empty());
284        assert_eq!(list.len(), 4);
285        let mut set: FallbackState = (&list).into();
286
287        // inspect the generated set
288        assert_eq!(set.fallbacks.len(), 4);
289        assert_eq!(
290            set.fallbacks[0].cmp_by_relay_ids(&set.fallbacks[1]),
291            O::Less
292        );
293        assert_eq!(
294            set.fallbacks[1].cmp_by_relay_ids(&set.fallbacks[2]),
295            O::Less
296        );
297        assert_eq!(
298            set.fallbacks[2].cmp_by_relay_ids(&set.fallbacks[3]),
299            O::Less
300        );
301
302        // use the constructed set a little.
303        for fb in fbs.iter() {
304            let id = FallbackId::from_relay_ids(fb);
305            assert_eq!(set.get_mut(&id).unwrap().cmp_by_relay_ids(&id), O::Equal);
306        }
307        assert!(set.get_mut(&id_other).is_none());
308
309        // Now try an input set with duplicates.
310        let mut redundant_fbs = fbs.clone();
311        redundant_fbs.extend(fbs.clone());
312        redundant_fbs.extend(fbs[0..2].iter().map(Clone::clone));
313        redundant_fbs[..].shuffle(&mut testing_rng());
314        let list2 = redundant_fbs.into();
315        assert_ne!(&list, &list2);
316        let set2: FallbackState = (&list2).into();
317
318        // It should have the same elements, in the same order.
319        assert_eq!(set.fallbacks.len(), set2.fallbacks.len());
320        assert!(set
321            .fallbacks
322            .iter()
323            .zip(set2.fallbacks.iter())
324            .all(|(ent1, ent2)| ent1.same_relay_ids(ent2)));
325    }
326
327    #[test]
328    fn set_choose() {
329        dbg!("X");
330
331        let mut rng = testing_rng();
332        let fbs = vec![
333            rand_fb(&mut rng),
334            rand_fb(&mut rng),
335            rand_fb(&mut rng),
336            rand_fb(&mut rng),
337        ];
338        let list: FallbackList = fbs.into();
339        let mut set: FallbackState = (&list).into();
340        let filter = crate::GuardFilter::unfiltered();
341
342        let mut counts = [0_usize; 4];
343        let now = Instant::now();
344        dbg!("A");
345        fn lookup_idx(set: &FallbackState, id: &impl HasRelayIds) -> Option<usize> {
346            set.fallbacks
347                .binary_search_by(|ent| ent.fallback.cmp_by_relay_ids(id))
348                .ok()
349        }
350        // Basic case: everybody is up.
351        for _ in 0..100 {
352            let fb = set.choose(&mut rng, now, &filter).unwrap();
353            let idx = lookup_idx(&set, fb).unwrap();
354            counts[idx] += 1;
355        }
356        dbg!("B");
357        assert!(counts.iter().all(|v| *v > 0));
358
359        // Mark somebody down and make sure they don't get chosen.
360        let ids: Vec<_> = set
361            .fallbacks
362            .iter()
363            .map(|ent| FallbackId::from_relay_ids(&ent.fallback))
364            .collect();
365        set.note_failure(&ids[2], now);
366        counts = [0; 4];
367        for _ in 0..100 {
368            let fb = set.choose(&mut rng, now, &filter).unwrap();
369            let idx = lookup_idx(&set, fb).unwrap();
370            counts[idx] += 1;
371        }
372        assert_eq!(counts.iter().filter(|v| **v > 0).count(), 3);
373        assert_eq!(counts[2], 0);
374
375        // Mark everybody down; make sure we get the right error.
376        for id in ids.iter() {
377            set.note_failure(id, now);
378        }
379        assert!(matches!(
380            set.choose(&mut rng, now, &filter),
381            Err(PickGuardError::AllFallbacksDown { .. })
382        ));
383
384        // Construct an empty set; make sure we get the right error.
385        let empty_set = FallbackState::from(&FallbackList::from(vec![]));
386        assert!(matches!(
387            empty_set.choose(&mut rng, now, &filter),
388            Err(PickGuardError::NoCandidatesAvailable)
389        ));
390
391        // TODO: test restrictions and filters once they're implemented.
392    }
393
394    #[test]
395    fn test_status() {
396        let mut rng = testing_rng();
397        let fbs = vec![
398            rand_fb(&mut rng),
399            rand_fb(&mut rng),
400            rand_fb(&mut rng),
401            rand_fb(&mut rng),
402        ];
403        let list: FallbackList = fbs.clone().into();
404        let mut set: FallbackState = (&list).into();
405        let ids: Vec<_> = set
406            .fallbacks
407            .iter()
408            .map(|ent| FallbackId::from_relay_ids(&ent.fallback))
409            .collect();
410
411        let now = Instant::now();
412
413        // There's no "next retry time" when everybody's up.
414        assert!(set.next_retry().is_none());
415
416        // Mark somebody down; try accessors.
417        set.note_failure(&ids[3], now);
418        assert!(set.fallbacks[3].status.next_retriable().unwrap() > now);
419        assert!(!set.fallbacks[3].status.usable_at(now));
420        assert_eq!(set.next_retry(), set.fallbacks[3].status.next_retriable());
421
422        // Mark somebody else down; try accessors.
423        set.note_failure(&ids[0], now);
424        assert!(set.fallbacks[0].status.next_retriable().unwrap() > now);
425        assert!(!set.fallbacks[0].status.usable_at(now));
426        assert_eq!(
427            set.next_retry().unwrap(),
428            std::cmp::min(
429                set.fallbacks[0].status.next_retriable().unwrap(),
430                set.fallbacks[3].status.next_retriable().unwrap()
431            )
432        );
433
434        // Mark somebody as running; try accessors.
435        set.note_success(&ids[0]);
436        assert!(set.fallbacks[0].status.next_retriable().is_none());
437        assert!(set.fallbacks[0].status.usable_at(now));
438
439        // Make a new set with slightly different members; make sure that we can copy stuff successfully.
440        let mut fbs2: Vec<_> = fbs
441            .into_iter()
442            // (Remove the fallback with id==ids[2])
443            .filter(|fb| FallbackId::from_relay_ids(fb) != ids[2])
444            .collect();
445        // add 2 new ones.
446        let fbs_new = vec![rand_fb(&mut rng), rand_fb(&mut rng), rand_fb(&mut rng)];
447        fbs2.extend(fbs_new.clone());
448
449        let mut set2 = FallbackState::from(&FallbackList::from(fbs2.clone()));
450        set2.take_status_from(set); // consumes set.
451        assert_eq!(set2.fallbacks.len(), 6); // Started with 4, added 3, removed 1.
452
453        // Make sure that the status entries  are correctly copied.
454        assert!(set2.get_mut(&ids[0]).unwrap().status.usable_at(now));
455        assert!(set2.get_mut(&ids[1]).unwrap().status.usable_at(now));
456        assert!(set2.get_mut(&ids[2]).is_none());
457        assert!(!set2.get_mut(&ids[3]).unwrap().status.usable_at(now));
458
459        // Make sure that the new fbs are there.
460        for new_fb in fbs_new {
461            assert!(set2
462                .get_mut(&FallbackId::from_relay_ids(&new_fb))
463                .unwrap()
464                .status
465                .usable_at(now));
466        }
467    }
468}
tor_guardmgr/fallback/set.rs

tor_guardmgr/fallback/
set.rs
