Skip to main content

zakura_network/
address_book.rs

1//! The `AddressBook` manages information about what peers exist, when they were
2//! seen, and what services they provide.
3
4use std::{
5    cmp::Reverse,
6    collections::HashMap,
7    net::{IpAddr, SocketAddr},
8    sync::{Arc, Mutex},
9    time::Instant,
10};
11
12use chrono::Utc;
13use indexmap::IndexMap;
14use ordered_map::OrderedMap;
15use tokio::sync::watch;
16use tracing::Span;
17
18use zakura_chain::{parameters::Network, serialization::DateTime32};
19
20use crate::{
21    constants::{self, ADDR_RESPONSE_LIMIT_DENOMINATOR, MAX_ADDRS_IN_MESSAGE},
22    meta_addr::MetaAddrChange,
23    protocol::external::{canonical_peer_addr, canonical_socket_addr},
24    types::{MetaAddr, PeerServices},
25    AddressBookPeers, PeerAddrState, PeerSocketAddr,
26};
27
28#[cfg(test)]
29mod tests;
30
31/// A database of peer listener addresses, their advertised services, and
32/// information on when they were last seen.
33///
34/// # Security
35///
36/// Address book state must be based on outbound connections to peers.
37///
38/// If the address book is updated incorrectly:
39/// - malicious peers can interfere with other peers' `AddressBook` state,
40///   or
41/// - Zebra can advertise unreachable addresses to its own peers.
42///
43/// ## Adding Addresses
44///
45/// The address book should only contain Zcash listener port addresses from peers
46/// on the configured network. These addresses can come from:
47/// - DNS seeders
48/// - addresses gossiped by other peers
49/// - the canonical address (`Version.address_from`) provided by each peer,
50///   particularly peers on inbound connections.
51///
52/// The remote addresses of inbound connections must not be added to the address
53/// book, because they contain ephemeral outbound ports, not listener ports.
54///
55/// Isolated connections must not add addresses or update the address book.
56///
57/// ## Updating Address State
58///
59/// Updates to address state must be based on outbound connections to peers.
60///
61/// Updates must not be based on:
62/// - the remote addresses of inbound connections, or
63/// - the canonical address of any connection.
64#[derive(Debug)]
65pub struct AddressBook {
66    /// Peer listener addresses, suitable for outbound connections,
67    /// in connection attempt order.
68    ///
69    /// Some peers in this list might have open outbound or inbound connections.
70    ///
71    /// We reverse the comparison order, because the standard library
72    /// ([`BTreeMap`](std::collections::BTreeMap)) sorts in ascending order, but
73    /// [`OrderedMap`] sorts in descending order.
74    by_addr: OrderedMap<PeerSocketAddr, MetaAddr, Reverse<MetaAddr>>,
75
76    /// The address with a last_connection_state of [`PeerAddrState::Responded`] and
77    /// the most recent `last_response` time by IP.
78    ///
79    /// This is used to avoid initiating outbound connections past [`Config::max_connections_per_ip`](crate::config::Config), and
80    /// currently only supports a `max_connections_per_ip` of 1, and must be `None` when used with a greater `max_connections_per_ip`.
81    // TODO: Replace with `by_ip: HashMap<IpAddr, BTreeMap<DateTime32, MetaAddr>>` to support configured `max_connections_per_ip` greater than 1
82    most_recent_by_ip: Option<HashMap<IpAddr, MetaAddr>>,
83
84    /// A list of banned addresses, with the time they were banned.
85    bans_by_ip: Arc<IndexMap<IpAddr, Instant>>,
86
87    /// The local listener address.
88    local_listener: SocketAddr,
89
90    /// The services advertised for our own [`local_listener`](Self::local_listener)
91    /// address when it is gossiped to peers.
92    ///
93    /// This must match the services advertised during the handshake, so a node
94    /// that does not advertise [`PeerServices::NODE_NETWORK`] (for example, a
95    /// pruned node) does not gossip itself as a full node.
96    local_listener_services: PeerServices,
97
98    /// The configured Zcash network.
99    network: Network,
100
101    /// The maximum number of addresses in the address book.
102    ///
103    /// Always set to [`MAX_ADDRS_IN_ADDRESS_BOOK`](constants::MAX_ADDRS_IN_ADDRESS_BOOK),
104    /// in release builds. Lower values are used during testing.
105    addr_limit: usize,
106
107    /// The span for operations on this address book.
108    span: Span,
109
110    /// A channel used to send the latest address book metrics.
111    address_metrics_tx: watch::Sender<AddressMetrics>,
112
113    /// The last time we logged a message about the address metrics.
114    last_address_log: Option<Instant>,
115}
116
117/// Metrics about the states of the addresses in an [`AddressBook`].
118#[derive(Copy, Clone, Debug, Default, PartialEq, Eq, Hash)]
119pub struct AddressMetrics {
120    /// The number of addresses in the `Responded` state.
121    pub responded: usize,
122
123    /// The number of addresses in the `NeverAttemptedGossiped` state.
124    pub never_attempted_gossiped: usize,
125
126    /// The number of addresses in the `Failed` state.
127    pub failed: usize,
128
129    /// The number of addresses in the `AttemptPending` state.
130    pub attempt_pending: usize,
131
132    /// The number of `Responded` addresses within the liveness limit.
133    pub recently_live: usize,
134
135    /// The number of `Responded` addresses outside the liveness limit.
136    pub recently_stopped_responding: usize,
137
138    /// The number of addresses in the address book, regardless of their states.
139    pub num_addresses: usize,
140
141    /// The maximum number of addresses in the address book.
142    pub address_limit: usize,
143}
144
145#[allow(clippy::len_without_is_empty)]
146impl AddressBook {
147    /// Construct an [`AddressBook`] with the given `local_listener` on `network`.
148    ///
149    /// Uses the supplied [`tracing::Span`] for address book operations.
150    pub fn new(
151        local_listener: SocketAddr,
152        network: &Network,
153        max_connections_per_ip: usize,
154        span: Span,
155    ) -> AddressBook {
156        let constructor_span = span.clone();
157        let _guard = constructor_span.enter();
158
159        let instant_now = Instant::now();
160        let chrono_now = Utc::now();
161
162        // The default value is correct for an empty address book,
163        // and it gets replaced by `update_metrics` anyway.
164        let (address_metrics_tx, _address_metrics_rx) = watch::channel(AddressMetrics::default());
165
166        // Avoid initiating outbound handshakes when max_connections_per_ip is 1.
167        let should_limit_outbound_conns_per_ip = max_connections_per_ip == 1;
168        let mut new_book = AddressBook {
169            by_addr: OrderedMap::new(|meta_addr| Reverse(*meta_addr)),
170            local_listener: canonical_socket_addr(local_listener),
171            // Default to full-node services; callers that advertise different
172            // services set them with `with_local_listener_services`.
173            local_listener_services: PeerServices::NODE_NETWORK,
174            network: network.clone(),
175            addr_limit: constants::MAX_ADDRS_IN_ADDRESS_BOOK,
176            span,
177            address_metrics_tx,
178            last_address_log: None,
179            most_recent_by_ip: should_limit_outbound_conns_per_ip.then(HashMap::new),
180            bans_by_ip: Default::default(),
181        };
182
183        new_book.update_metrics(instant_now, chrono_now);
184        new_book
185    }
186
187    /// Sets the services advertised for our own gossiped listener address.
188    ///
189    /// These must match the services advertised during the handshake.
190    #[must_use]
191    pub fn with_local_listener_services(mut self, services: PeerServices) -> Self {
192        self.local_listener_services = services;
193        self
194    }
195
196    /// Construct an [`AddressBook`] with the given `local_listener`, `network`,
197    /// `addr_limit`, [`tracing::Span`], and addresses.
198    ///
199    /// `addr_limit` is enforced by this method, and by [`AddressBook::update`].
200    ///
201    /// If there are multiple [`MetaAddr`]s with the same address,
202    /// an arbitrary address is inserted into the address book,
203    /// and the rest are dropped.
204    ///
205    /// This constructor can be used to break address book invariants,
206    /// so it should only be used in tests.
207    #[cfg(any(test, feature = "proptest-impl"))]
208    pub fn new_with_addrs(
209        local_listener: SocketAddr,
210        network: &Network,
211        max_connections_per_ip: usize,
212        addr_limit: usize,
213        span: Span,
214        addrs: impl IntoIterator<Item = MetaAddr>,
215    ) -> AddressBook {
216        let constructor_span = span.clone();
217        let _guard = constructor_span.enter();
218
219        let instant_now = Instant::now();
220        let chrono_now = Utc::now();
221
222        // The maximum number of addresses should be always greater than 0
223        assert!(addr_limit > 0);
224
225        let mut new_book = AddressBook::new(local_listener, network, max_connections_per_ip, span);
226        new_book.addr_limit = addr_limit;
227
228        let addrs = addrs
229            .into_iter()
230            .map(|mut meta_addr| {
231                meta_addr.addr = canonical_peer_addr(meta_addr.addr);
232                meta_addr
233            })
234            .filter(|meta_addr| meta_addr.address_is_valid_for_outbound(network))
235            .map(|meta_addr| (meta_addr.addr, meta_addr));
236
237        for (socket_addr, meta_addr) in addrs {
238            // overwrite any duplicate addresses
239            new_book.by_addr.insert(socket_addr, meta_addr);
240            // Add the address to `most_recent_by_ip` if it has responded
241            if new_book.should_update_most_recent_by_ip(meta_addr) {
242                new_book
243                    .most_recent_by_ip
244                    .as_mut()
245                    .expect("should be some when should_update_most_recent_by_ip is true")
246                    .insert(socket_addr.ip(), meta_addr);
247            }
248            // exit as soon as we get enough addresses
249            if new_book.by_addr.len() >= addr_limit {
250                break;
251            }
252        }
253
254        new_book.update_metrics(instant_now, chrono_now);
255        new_book
256    }
257
258    /// Return a watch channel for the address book metrics.
259    ///
260    /// The metrics in the watch channel are only updated when the address book updates,
261    /// so they can be significantly outdated if Zebra is disconnected or hung.
262    ///
263    /// The current metrics value is marked as seen.
264    /// So `Receiver::changed` will only return after the next address book update.
265    pub fn address_metrics_watcher(&self) -> watch::Receiver<AddressMetrics> {
266        self.address_metrics_tx.subscribe()
267    }
268
269    /// Set the local listener address. Only for use in tests.
270    #[cfg(any(test, feature = "proptest-impl"))]
271    pub fn set_local_listener(&mut self, addr: SocketAddr) {
272        self.local_listener = addr;
273    }
274
275    /// Get the local listener address.
276    ///
277    /// This address contains minimal state, but it is not sanitized.
278    pub fn local_listener_meta_addr(&self, now: chrono::DateTime<Utc>) -> MetaAddr {
279        let now: DateTime32 = now.try_into().expect("will succeed until 2038");
280
281        MetaAddr::new_local_listener_change(self.local_listener, self.local_listener_services)
282            .local_listener_into_new_meta_addr(now)
283    }
284
285    /// Get the local listener [`SocketAddr`].
286    pub fn local_listener_socket_addr(&self) -> SocketAddr {
287        self.local_listener
288    }
289
290    /// Get the active addresses in `self` in random order with sanitized timestamps,
291    /// including our local listener address.
292    ///
293    /// Limited to the number of peer addresses Zebra should give out per `GetAddr` request.
294    pub fn fresh_get_addr_response(&self) -> Vec<MetaAddr> {
295        let now = Utc::now();
296        let mut peers = self.sanitized(now);
297        let address_limit = peers.len().div_ceil(ADDR_RESPONSE_LIMIT_DENOMINATOR);
298        peers.truncate(MAX_ADDRS_IN_MESSAGE.min(address_limit));
299
300        peers
301    }
302
303    /// Get the active addresses in `self` in random order with sanitized timestamps,
304    /// including our local listener address.
305    pub(crate) fn sanitized(&self, now: chrono::DateTime<Utc>) -> Vec<MetaAddr> {
306        use rand::seq::SliceRandom;
307        let _guard = self.span.enter();
308
309        let mut peers = self.by_addr.clone();
310
311        // Unconditionally add our local listener address to the advertised peers,
312        // to replace any self-connection failures. The address book and change
313        // constructors make sure that the SocketAddr is canonical.
314        let local_listener = self.local_listener_meta_addr(now);
315        peers.insert(local_listener.addr, local_listener);
316
317        // Then sanitize and shuffle
318        let mut peers: Vec<MetaAddr> = peers
319            .descending_values()
320            .filter_map(|meta_addr| meta_addr.sanitize(&self.network))
321            // # Security
322            //
323            // Remove peers that:
324            //   - last responded more than three hours ago, or
325            //   - haven't responded yet but were reported last seen more than three hours ago
326            //
327            // This prevents Zebra from gossiping nodes that are likely unreachable. Gossiping such
328            // nodes impacts the network health, because connection attempts end up being wasted on
329            // peers that are less likely to respond.
330            .filter(|addr| addr.is_active_for_gossip(now))
331            .collect();
332
333        peers.shuffle(&mut rand::thread_rng());
334
335        peers
336    }
337
338    /// Get the active addresses in `self`, in preferred caching order,
339    /// excluding our local listener address.
340    pub fn cacheable(&self, now: chrono::DateTime<Utc>) -> Vec<MetaAddr> {
341        let _guard = self.span.enter();
342
343        let peers = self.by_addr.clone();
344
345        // Get peers in preferred order, then keep the recently active ones
346        peers
347            .descending_values()
348            // # Security
349            //
350            // Remove peers that:
351            //   - last responded more than three hours ago, or
352            //   - haven't responded yet but were reported last seen more than three hours ago
353            //
354            // This prevents Zebra from caching nodes that are likely unreachable,
355            // which improves startup time and reliability.
356            .filter(|addr| addr.is_active_for_gossip(now))
357            .cloned()
358            .collect()
359    }
360
361    /// Look up `addr` in the address book, and return its [`MetaAddr`].
362    ///
363    /// Converts `addr` to a canonical address before looking it up.
364    pub fn get(&mut self, addr: PeerSocketAddr) -> Option<MetaAddr> {
365        let addr = canonical_peer_addr(*addr);
366
367        // Unfortunately, `OrderedMap` doesn't implement `get`.
368        let meta_addr = self.by_addr.remove(&addr);
369
370        if let Some(meta_addr) = meta_addr {
371            self.by_addr.insert(addr, meta_addr);
372        }
373
374        meta_addr
375    }
376
377    /// Returns true if `updated` needs to be applied to the recent outbound peer connection IP cache.
378    ///
379    /// Checks if there are no existing entries in the address book with this IP,
380    /// or if `updated` has a more recent `last_response` requiring the outbound connector to wait
381    /// longer before initiating handshakes with peers at this IP.
382    ///
383    /// This code only needs to check a single cache entry, rather than the entire address book,
384    /// because other code maintains these invariants:
385    /// - `last_response` times for an entry can only increase.
386    /// - this is the only field checked by `has_connection_recently_responded()`
387    ///
388    /// See [`AddressBook::is_ready_for_connection_attempt_with_ip`] for more details.
389    fn should_update_most_recent_by_ip(&self, updated: MetaAddr) -> bool {
390        let Some(most_recent_by_ip) = self.most_recent_by_ip.as_ref() else {
391            return false;
392        };
393
394        if let Some(previous) = most_recent_by_ip.get(&updated.addr.ip()) {
395            updated.last_connection_state == PeerAddrState::Responded
396                && updated.last_response() > previous.last_response()
397        } else {
398            updated.last_connection_state == PeerAddrState::Responded
399        }
400    }
401
402    /// Returns true if `addr` is the latest entry for its IP, which is stored in `most_recent_by_ip`.
403    /// The entry is checked for an exact match to the IP and port of `addr`.
404    fn should_remove_most_recent_by_ip(&self, addr: PeerSocketAddr) -> bool {
405        let Some(most_recent_by_ip) = self.most_recent_by_ip.as_ref() else {
406            return false;
407        };
408
409        if let Some(previous) = most_recent_by_ip.get(&addr.ip()) {
410            previous.addr == addr
411        } else {
412            false
413        }
414    }
415
416    /// Apply `change` to the address book, returning the updated `MetaAddr`,
417    /// if the change was valid.
418    ///
419    /// # Correctness
420    ///
421    /// All changes should go through `update`, so that the address book
422    /// only contains valid outbound addresses.
423    ///
424    /// Change addresses must be canonical `PeerSocketAddr`s. This makes sure that
425    /// each address book entry has a unique IP address.
426    ///
427    /// # Security
428    ///
429    /// This function must apply every attempted, responded, and failed change
430    /// to the address book. This prevents rapid reconnections to the same peer.
431    ///
432    /// As an exception, this function can ignore all changes for specific
433    /// [`PeerSocketAddr`]s. Ignored addresses will never be used to connect to
434    /// peers.
435    #[allow(clippy::unwrap_in_result)]
436    pub fn update(&mut self, change: MetaAddrChange) -> Option<MetaAddr> {
437        if self.bans_by_ip.contains_key(&change.addr().ip()) {
438            tracing::warn!(
439                ?change,
440                "attempted to add a banned peer addr to address book"
441            );
442            return None;
443        }
444
445        let previous = self.get(change.addr());
446
447        let _guard = self.span.enter();
448
449        let instant_now = Instant::now();
450        let chrono_now = Utc::now();
451
452        let updated = change.apply_to_meta_addr(previous, instant_now, chrono_now);
453
454        trace!(
455            ?change,
456            ?updated,
457            ?previous,
458            total_peers = self.by_addr.len(),
459            recent_peers = self.recently_live_peers(chrono_now).len(),
460            "calculated updated address book entry",
461        );
462
463        if let Some(updated) = updated {
464            if updated.misbehavior() >= constants::MAX_PEER_MISBEHAVIOR_SCORE {
465                // Ban and skip outbound connections with excessively misbehaving peers.
466                let banned_ip = updated.addr.ip();
467                let bans_by_ip = Arc::make_mut(&mut self.bans_by_ip);
468
469                bans_by_ip.insert(banned_ip, Instant::now());
470                if bans_by_ip.len() > constants::MAX_BANNED_IPS {
471                    // Remove the oldest banned IP from the address book.
472                    bans_by_ip.shift_remove_index(0);
473                }
474
475                // `most_recent_by_ip` is only populated when
476                // `max_connections_per_ip == 1`. The ban path runs for any
477                // configured value, so we must guard the optional cache rather
478                // than unwrap it.
479                if let Some(most_recent_by_ip) = self.most_recent_by_ip.as_mut() {
480                    most_recent_by_ip.remove(&banned_ip);
481                }
482
483                let banned_addrs: Vec<_> = self
484                    .by_addr
485                    .descending_keys()
486                    .skip_while(|addr| addr.ip() != banned_ip)
487                    .take_while(|addr| addr.ip() == banned_ip)
488                    .cloned()
489                    .collect();
490
491                for addr in banned_addrs {
492                    self.by_addr.remove(&addr);
493                }
494
495                warn!(
496                    ?updated,
497                    total_peers = self.by_addr.len(),
498                    recent_peers = self.recently_live_peers(chrono_now).len(),
499                    "banned ip and removed banned peer addresses from address book",
500                );
501
502                return None;
503            }
504
505            // Ignore invalid outbound addresses.
506            // (Inbound connections can be monitored via Zebra's metrics.)
507            if !updated.address_is_valid_for_outbound(&self.network) {
508                return None;
509            }
510
511            // Ignore invalid outbound services and other info,
512            // but only if the peer has never been attempted.
513            //
514            // Otherwise, if we got the info directly from the peer,
515            // store it in the address book, so we know not to reconnect.
516            if !updated.last_known_info_is_valid_for_outbound(&self.network)
517                && updated.last_connection_state.is_never_attempted()
518            {
519                return None;
520            }
521
522            self.by_addr.insert(updated.addr, updated);
523
524            // Add the address to `most_recent_by_ip` if it sent the most recent
525            // response Zebra has received from this IP.
526            if self.should_update_most_recent_by_ip(updated) {
527                self.most_recent_by_ip
528                    .as_mut()
529                    .expect("should be some when should_update_most_recent_by_ip is true")
530                    .insert(updated.addr.ip(), updated);
531            }
532
533            debug!(
534                ?change,
535                ?updated,
536                ?previous,
537                total_peers = self.by_addr.len(),
538                recent_peers = self.recently_live_peers(chrono_now).len(),
539                "updated address book entry",
540            );
541
542            // Security: Limit the number of peers in the address book.
543            //
544            // We only delete outdated peers when we have too many peers.
545            // If we deleted them as soon as they became too old,
546            // then other peers could re-insert them into the address book.
547            // And we would start connecting to those outdated peers again,
548            // ignoring the age limit in [`MetaAddr::is_probably_reachable`].
549            while self.by_addr.len() > self.addr_limit {
550                let surplus_peer = self
551                    .peers()
552                    .next_back()
553                    .expect("just checked there is at least one peer");
554
555                self.by_addr.remove(&surplus_peer.addr);
556
557                // Check if this surplus peer's addr matches that in `most_recent_by_ip`
558                // for this the surplus peer's ip to remove it there as well.
559                if self.should_remove_most_recent_by_ip(surplus_peer.addr) {
560                    self.most_recent_by_ip
561                        .as_mut()
562                        .expect("should be some when should_remove_most_recent_by_ip is true")
563                        .remove(&surplus_peer.addr.ip());
564                }
565
566                debug!(
567                    surplus = ?surplus_peer,
568                    ?updated,
569                    total_peers = self.by_addr.len(),
570                    recent_peers = self.recently_live_peers(chrono_now).len(),
571                    "removed surplus address book entry",
572                );
573            }
574
575            assert!(self.len() <= self.addr_limit);
576
577            std::mem::drop(_guard);
578            self.update_metrics(instant_now, chrono_now);
579        }
580
581        updated
582    }
583
584    /// Removes the entry with `addr`, returning it if it exists
585    ///
586    /// # Note
587    ///
588    /// All address removals should go through `take`, so that the address
589    /// book metrics are accurate.
590    #[allow(dead_code)]
591    fn take(&mut self, removed_addr: PeerSocketAddr) -> Option<MetaAddr> {
592        let _guard = self.span.enter();
593
594        let instant_now = Instant::now();
595        let chrono_now = Utc::now();
596
597        trace!(
598            ?removed_addr,
599            total_peers = self.by_addr.len(),
600            recent_peers = self.recently_live_peers(chrono_now).len(),
601        );
602
603        if let Some(entry) = self.by_addr.remove(&removed_addr) {
604            // Check if this surplus peer's addr matches that in `most_recent_by_ip`
605            // for this the surplus peer's ip to remove it there as well.
606            if self.should_remove_most_recent_by_ip(entry.addr) {
607                if let Some(most_recent_by_ip) = self.most_recent_by_ip.as_mut() {
608                    most_recent_by_ip.remove(&entry.addr.ip());
609                }
610            }
611
612            std::mem::drop(_guard);
613            self.update_metrics(instant_now, chrono_now);
614            Some(entry)
615        } else {
616            None
617        }
618    }
619
620    /// Returns true if the given [`PeerSocketAddr`] is pending a reconnection
621    /// attempt.
622    pub fn pending_reconnection_addr(&mut self, addr: PeerSocketAddr) -> bool {
623        let meta_addr = self.get(addr);
624
625        let _guard = self.span.enter();
626        match meta_addr {
627            None => false,
628            Some(peer) => peer.last_connection_state == PeerAddrState::AttemptPending,
629        }
630    }
631
632    /// Return an iterator over all peers.
633    ///
634    /// Returns peers in reconnection attempt order, including recently connected peers.
635    pub fn peers(&'_ self) -> impl DoubleEndedIterator<Item = MetaAddr> + '_ {
636        let _guard = self.span.enter();
637        self.by_addr.descending_values().cloned()
638    }
639
640    /// Is this IP ready for a new outbound connection attempt?
641    /// Checks if the outbound connection with the most recent response at this IP has recently responded.
642    ///
643    /// Note: last_response times may remain live for a long time if the local clock is changed to an earlier time.
644    fn is_ready_for_connection_attempt_with_ip(
645        &self,
646        ip: &IpAddr,
647        chrono_now: chrono::DateTime<Utc>,
648    ) -> bool {
649        let Some(most_recent_by_ip) = self.most_recent_by_ip.as_ref() else {
650            // if we're not checking IPs, any connection is allowed
651            return true;
652        };
653        let Some(same_ip_peer) = most_recent_by_ip.get(ip) else {
654            // If there's no entry for this IP, any connection is allowed
655            return true;
656        };
657        !same_ip_peer.has_connection_recently_responded(chrono_now)
658    }
659
660    /// Return an iterator over peers that are due for a reconnection attempt,
661    /// in reconnection attempt order.
662    pub fn reconnection_peers(
663        &'_ self,
664        instant_now: Instant,
665        chrono_now: chrono::DateTime<Utc>,
666    ) -> impl DoubleEndedIterator<Item = MetaAddr> + '_ {
667        let _guard = self.span.enter();
668
669        // Skip live peers, and peers pending a reconnect attempt.
670        // The peers are already stored in sorted order.
671        self.by_addr
672            .descending_values()
673            .filter(move |peer| {
674                peer.is_ready_for_connection_attempt(instant_now, chrono_now, &self.network)
675                    && self.is_ready_for_connection_attempt_with_ip(&peer.addr.ip(), chrono_now)
676            })
677            .cloned()
678    }
679
680    /// Return an iterator over all the peers in `state`,
681    /// in reconnection attempt order, including recently connected peers.
682    pub fn state_peers(
683        &'_ self,
684        state: PeerAddrState,
685    ) -> impl DoubleEndedIterator<Item = MetaAddr> + '_ {
686        let _guard = self.span.enter();
687
688        self.by_addr
689            .descending_values()
690            .filter(move |peer| peer.last_connection_state == state)
691            .cloned()
692    }
693
694    /// Return an iterator over peers that might be connected,
695    /// in reconnection attempt order.
696    pub fn maybe_connected_peers(
697        &'_ self,
698        instant_now: Instant,
699        chrono_now: chrono::DateTime<Utc>,
700    ) -> impl DoubleEndedIterator<Item = MetaAddr> + '_ {
701        let _guard = self.span.enter();
702
703        self.by_addr
704            .descending_values()
705            .filter(move |peer| {
706                !peer.is_ready_for_connection_attempt(instant_now, chrono_now, &self.network)
707            })
708            .cloned()
709    }
710
711    /// Returns banned IP addresses.
712    pub fn bans(&self) -> Arc<IndexMap<IpAddr, Instant>> {
713        self.bans_by_ip.clone()
714    }
715
716    /// Returns the number of entries in this address book.
717    pub fn len(&self) -> usize {
718        self.by_addr.len()
719    }
720
721    /// Returns metrics for the addresses in this address book.
722    /// Only for use in tests.
723    ///
724    /// # Correctness
725    ///
726    /// Use [`AddressBook::address_metrics_watcher`] in production code,
727    /// to avoid deadlocks.
728    #[cfg(test)]
729    pub fn address_metrics(&self, now: chrono::DateTime<Utc>) -> AddressMetrics {
730        self.address_metrics_internal(now)
731    }
732
733    /// Returns metrics for the addresses in this address book.
734    ///
735    /// # Correctness
736    ///
737    /// External callers should use [`AddressBook::address_metrics_watcher`]
738    /// in production code, to avoid deadlocks.
739    /// (Using the watch channel receiver does not lock the address book mutex.)
740    fn address_metrics_internal(&self, now: chrono::DateTime<Utc>) -> AddressMetrics {
741        let responded = self.state_peers(PeerAddrState::Responded).count();
742        let never_attempted_gossiped = self
743            .state_peers(PeerAddrState::NeverAttemptedGossiped)
744            .count();
745        let failed = self.state_peers(PeerAddrState::Failed).count();
746        let attempt_pending = self.state_peers(PeerAddrState::AttemptPending).count();
747
748        let recently_live = self.recently_live_peers(now).len();
749        let recently_stopped_responding = responded
750            .checked_sub(recently_live)
751            .expect("all recently live peers must have responded");
752
753        let num_addresses = self.len();
754
755        AddressMetrics {
756            responded,
757            never_attempted_gossiped,
758            failed,
759            attempt_pending,
760            recently_live,
761            recently_stopped_responding,
762            num_addresses,
763            address_limit: self.addr_limit,
764        }
765    }
766
767    /// Update the metrics for this address book.
768    fn update_metrics(&mut self, instant_now: Instant, chrono_now: chrono::DateTime<Utc>) {
769        let _guard = self.span.enter();
770
771        let m = self.address_metrics_internal(chrono_now);
772
773        // Ignore errors: we don't care if any receivers are listening.
774        let _ = self.address_metrics_tx.send(m);
775
776        // TODO: rename to address_book.[state_name]
777        metrics::gauge!("candidate_set.responded").set(m.responded as f64);
778        metrics::gauge!("candidate_set.gossiped").set(m.never_attempted_gossiped as f64);
779        metrics::gauge!("candidate_set.failed").set(m.failed as f64);
780        metrics::gauge!("candidate_set.pending").set(m.attempt_pending as f64);
781
782        // TODO: rename to address_book.responded.recently_live
783        metrics::gauge!("candidate_set.recently_live").set(m.recently_live as f64);
784        // TODO: rename to address_book.responded.stopped_responding
785        metrics::gauge!("candidate_set.disconnected").set(m.recently_stopped_responding as f64);
786
787        std::mem::drop(_guard);
788        self.log_metrics(&m, instant_now);
789    }
790
791    /// Log metrics for this address book
792    fn log_metrics(&mut self, m: &AddressMetrics, now: Instant) {
793        let _guard = self.span.enter();
794
795        trace!(
796            address_metrics = ?m,
797        );
798
799        if m.responded > 0 {
800            return;
801        }
802
803        // These logs are designed to be human-readable in a terminal, at the
804        // default Zebra log level. If you need to know address states for
805        // every request, use the trace-level logs, or the metrics exporter.
806        if let Some(last_address_log) = self.last_address_log {
807            // Avoid duplicate address logs
808            if now.saturating_duration_since(last_address_log).as_secs() < 60 {
809                return;
810            }
811        } else {
812            // Suppress initial logs until the peer set has started up.
813            // There can be multiple address changes before the first peer has
814            // responded.
815            self.last_address_log = Some(now);
816            return;
817        }
818
819        self.last_address_log = Some(now);
820        // if all peers have failed
821        if m.responded + m.attempt_pending + m.never_attempted_gossiped == 0 {
822            warn!(
823                address_metrics = ?m,
824                "all peer addresses have failed. Hint: check your network connection"
825            );
826        } else {
827            info!(
828                address_metrics = ?m,
829                "no active peer connections: trying gossiped addresses"
830            );
831        }
832    }
833}
834
835impl AddressBookPeers for AddressBook {
836    fn recently_live_peers(&self, now: chrono::DateTime<Utc>) -> Vec<MetaAddr> {
837        let _guard = self.span.enter();
838
839        self.by_addr
840            .descending_values()
841            .filter(|peer| peer.was_recently_live(now))
842            .cloned()
843            .collect()
844    }
845
846    fn add_peer(&mut self, peer: PeerSocketAddr) -> bool {
847        if self.get(peer).is_some() {
848            // Peer already exists in the address book, so we don't need to add it again.
849            return false;
850        }
851        self.update(MetaAddr::new_initial_peer(peer)).is_some()
852    }
853}
854
855impl AddressBookPeers for Arc<Mutex<AddressBook>> {
856    fn recently_live_peers(&self, now: chrono::DateTime<Utc>) -> Vec<MetaAddr> {
857        self.lock()
858            .expect("panic in a previous thread that was holding the mutex")
859            .recently_live_peers(now)
860    }
861
862    fn add_peer(&mut self, peer: PeerSocketAddr) -> bool {
863        self.lock()
864            .expect("panic in a previous thread that was holding the mutex")
865            .add_peer(peer)
866    }
867}
868
869impl Extend<MetaAddrChange> for AddressBook {
870    fn extend<T>(&mut self, iter: T)
871    where
872        T: IntoIterator<Item = MetaAddrChange>,
873    {
874        for change in iter.into_iter() {
875            self.update(change);
876        }
877    }
878}
879
880impl Clone for AddressBook {
881    /// Clone the addresses, address limit, local listener address, and span.
882    ///
883    /// Cloned address books have a separate metrics struct watch channel, and an empty last address log.
884    ///
885    /// All address books update the same prometheus metrics.
886    fn clone(&self) -> AddressBook {
887        // The existing metrics might be outdated, but we avoid calling `update_metrics`,
888        // so we don't overwrite the prometheus metrics from the main address book.
889        let (address_metrics_tx, _address_metrics_rx) =
890            watch::channel(*self.address_metrics_tx.borrow());
891
892        AddressBook {
893            by_addr: self.by_addr.clone(),
894            local_listener: self.local_listener,
895            local_listener_services: self.local_listener_services,
896            network: self.network.clone(),
897            addr_limit: self.addr_limit,
898            span: self.span.clone(),
899            address_metrics_tx,
900            last_address_log: None,
901            most_recent_by_ip: self.most_recent_by_ip.clone(),
902            bans_by_ip: self.bans_by_ip.clone(),
903        }
904    }
905}