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