Skip to main content

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