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}