1use crate::{
7 Error,
8 peer_gossip::{is_public, validate_gossip_address},
9 peer_protocol::GossipPeer,
10};
11use bitcoin::BlockHash;
12use libp2p::{Multiaddr, PeerId, multiaddr::Protocol};
13use rand::seq::SliceRandom;
14use serde::{Deserialize, Serialize};
15use std::{fs, io::ErrorKind, net::IpAddr, path::Path};
16use urma_node_client::model::BookEntry as ListedEntry;
17
18#[derive(Clone, Debug)]
21pub struct PeerAddress {
22 peer: PeerId,
23 address: Multiaddr,
24}
25
26impl PeerAddress {
27 pub fn new(peer: PeerId, address: Multiaddr) -> Result<Self, Error> {
31 let mut protocols = address.iter();
32 let valid_ip = match protocols.next() {
33 Some(Protocol::Ip4(ip)) => usable(IpAddr::V4(ip)),
34 Some(Protocol::Ip6(ip)) => usable(IpAddr::V6(ip)),
35 _ => false,
36 };
37 if !valid_ip
38 || !matches!(protocols.next(), Some(Protocol::Tcp(port)) if port != 0)
39 || !matches!(protocols.next(), Some(Protocol::P2p(id)) if id == peer)
40 || protocols.next().is_some()
41 {
42 return Err(Error::Invalid(
43 "peer address requires a usable IP, nonzero TCP port and terminal matching peer ID"
44 .into(),
45 ));
46 }
47 Ok(Self { peer, address })
48 }
49
50 pub fn peer_id(&self) -> PeerId {
51 self.peer
52 }
53
54 pub fn address(&self) -> &Multiaddr {
55 &self.address
56 }
57}
58
59pub use urma_node_client::model::BookOrigin as Origin;
60
61#[derive(Clone, Debug, PartialEq, Eq)]
64pub struct BookEntry {
65 pub peer_id: Option<PeerId>,
66 pub addrs: Vec<Multiaddr>,
67 pub origin: Origin,
68 pub last_success: Option<u64>,
69 pub failures: u32,
70 pub banned_until: Option<u64>,
71}
72
73#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize)]
76pub struct BookStats {
77 pub seeds: u64,
78 pub book: u64,
79 pub reached: u64,
80 pub reached_ever: u64,
81 pub gossip: u64,
82}
83
84#[derive(Clone, Copy, Debug, PartialEq, Eq)]
87enum Learned {
88 Operator,
89 Gossip,
90 Identify,
91}
92
93pub struct PeerBook {
94 chain_genesis: BlockHash,
95 entries: Vec<BookEntry>,
96}
97
98#[derive(Serialize, Deserialize)]
99#[serde(deny_unknown_fields)]
100struct SavedBook {
101 version: u32,
102 chain_genesis: String,
103 entries: Vec<SavedEntry>,
104}
105
106#[derive(Serialize, Deserialize)]
108#[serde(deny_unknown_fields)]
109struct SavedEntry {
110 peer_id: Option<String>,
111 addrs: Vec<String>,
112 origin: Origin,
113 last_success: Option<u64>,
114 failures: u32,
115 banned_until: Option<u64>,
116}
117
118impl From<SavedEntry> for ListedEntry {
119 fn from(saved: SavedEntry) -> Self {
120 let SavedEntry {
121 peer_id,
122 addrs,
123 origin,
124 last_success,
125 failures,
126 banned_until,
127 } = saved;
128 Self {
129 peer_id,
130 addrs,
131 origin,
132 last_success,
133 failures,
134 banned_until,
135 }
136 }
137}
138
139enum Host {
141 Ip(IpAddr),
142 Name,
143}
144
145impl PeerBook {
146 pub const CAPACITY: usize = 1024;
147 pub const ADDRESSES: usize = 4;
148 pub const GROUP_LIMIT: usize = 2;
150 const FILE_LIMIT: usize = 1024 * 1024;
151 const VERSION: u32 = 2;
152 const BACKOFF_SECONDS: u64 = 60;
153 const BACKOFF_CAP: u64 = 86_400;
154 pub const SEED_BACKOFF_CAP: u64 = 600;
157
158 pub fn empty(chain_genesis: BlockHash) -> Self {
159 Self {
160 chain_genesis,
161 entries: Vec::new(),
162 }
163 }
164
165 pub fn load(path: &Path, chain_genesis: BlockHash) -> Result<Self, Error> {
169 let mut book = Self::empty(chain_genesis);
170 let metadata = match fs::symlink_metadata(path) {
171 Ok(metadata) => metadata,
172 Err(error) if error.kind() == ErrorKind::NotFound => return Ok(book),
173 Err(error) => return Err(error.into()),
174 };
175 let limit = u64::try_from(Self::FILE_LIMIT).map_err(|e| Error::Invalid(e.to_string()))?;
176 if !metadata.is_file() || metadata.len() > limit {
177 return Err(Error::Invalid(
178 "peer book must be a regular file of at most 1 MiB".into(),
179 ));
180 }
181 let bytes = urma_io::read_regular(path, Self::FILE_LIMIT).map_err(io_source)?;
182 let saved: SavedBook = serde_json::from_slice(&bytes)?;
183 if saved.version != Self::VERSION {
184 return Err(Error::Invalid("unsupported peer book version".into()));
185 }
186 let saved_genesis: BlockHash = saved
187 .chain_genesis
188 .parse()
189 .map_err(|_| Error::Invalid("invalid peer book chain genesis".into()))?;
190 if saved_genesis != chain_genesis {
191 return Err(Error::Invalid("peer book belongs to another chain".into()));
192 }
193 if saved.entries.len() > Self::CAPACITY {
194 return Err(Error::Busy("peer book has too many entries".into()));
195 }
196 for saved in saved.entries {
197 let entry = restore(saved)?;
198 if book.overlaps(&entry) {
199 return Err(Error::Invalid("duplicate saved peer or address".into()));
200 }
201 book.entries.push(entry);
202 }
203 Ok(book)
204 }
205
206 pub fn save(&self, path: &Path) -> Result<(), Error> {
208 let mut entries: Vec<SavedEntry> = self.entries.iter().map(persist).collect();
209 entries.sort_by(|a, b| (&a.peer_id, &a.addrs).cmp(&(&b.peer_id, &b.addrs)));
210 let saved = SavedBook {
211 version: Self::VERSION,
212 chain_genesis: self.chain_genesis.to_string(),
213 entries,
214 };
215 let bytes = serde_json::to_vec(&saved)?;
216 if bytes.len() > Self::FILE_LIMIT {
217 return Err(Error::Invalid("peer book exceeds 1 MiB".into()));
218 }
219 urma_io::write_replace(path, &bytes).map_err(io_source)
220 }
221
222 pub fn entries(&self) -> &[BookEntry] {
223 &self.entries
224 }
225
226 pub fn add(&mut self, address: Multiaddr, origin: Origin) -> Result<bool, Error> {
232 let learned = match origin {
233 Origin::Gossip => Learned::Gossip,
234 Origin::Seed | Origin::Manual => Learned::Operator,
235 };
236 self.insert(address, origin, learned)
237 }
238
239 fn insert(
243 &mut self,
244 address: Multiaddr,
245 origin: Origin,
246 learned: Learned,
247 ) -> Result<bool, Error> {
248 let (address, peer) = split(address)?;
249 if origin == Origin::Gossip {
250 self.admit_gossip(&address)?;
251 }
252 if let Some(index) = self.position_of_address(&address) {
253 if learned != Learned::Operator {
254 return Ok(false);
255 }
256 let entry = &mut self.entries[index];
257 if peer.is_some() && entry.peer_id.is_some() && entry.peer_id != peer {
258 return Err(Error::Invalid("address already names another peer".into()));
259 }
260 let upgraded = origin > entry.origin || (entry.peer_id.is_none() && peer.is_some());
261 entry.origin = entry.origin.max(origin);
262 entry.peer_id = entry.peer_id.or(peer);
263 return Ok(upgraded);
264 }
265 if let Some(index) = peer.and_then(|peer| self.position_of_peer(peer)) {
266 if learned == Learned::Gossip {
267 return Ok(false);
268 }
269 let entry = &mut self.entries[index];
270 if entry.addrs.len() == Self::ADDRESSES {
271 return Err(Error::Busy("peer already has four addresses".into()));
272 }
273 entry.addrs.push(address);
274 entry.origin = entry.origin.max(origin);
275 return Ok(true);
276 }
277 self.make_room(origin)?;
278 self.entries.push(BookEntry {
279 peer_id: peer,
280 addrs: vec![address],
281 origin,
282 last_success: None,
283 failures: 0,
284 banned_until: None,
285 });
286 Ok(true)
287 }
288
289 pub fn merge(&mut self, peers: &[GossipPeer], local: PeerId, omitting: &[String]) -> usize {
293 self.learn(peers, local, omitting, Learned::Gossip)
294 }
295
296 pub fn merge_identified(
299 &mut self,
300 peers: &[GossipPeer],
301 local: PeerId,
302 omitting: &[String],
303 ) -> usize {
304 self.learn(peers, local, omitting, Learned::Identify)
305 }
306
307 fn learn(
308 &mut self,
309 peers: &[GossipPeer],
310 local: PeerId,
311 omitting: &[String],
312 learned: Learned,
313 ) -> usize {
314 let mut changed = 0;
315 for peer in peers {
316 let id = peer.peer_id;
317 if id == local || omitting.contains(&id.to_string()) {
318 continue;
319 }
320 let mut addresses = Vec::new();
321 for address in peer.addrs.iter().take(Self::ADDRESSES) {
322 let Ok((address, named)) = split(address.clone()) else {
323 continue;
324 };
325 if named.is_none_or(|named| named == id) && !addresses.contains(&address) {
326 addresses.push(address);
327 }
328 }
329 if self.position_of_peer(id).is_none() {
330 changed += self.insert_learned_peer(id, addresses);
331 continue;
332 }
333 if learned != Learned::Identify {
334 continue;
335 }
336 for address in addresses {
337 let address = address.with(Protocol::P2p(id));
338 if let Ok(true) = self.insert(address, Origin::Gossip, learned) {
339 changed += 1;
340 }
341 }
342 }
343 changed
344 }
345
346 fn insert_learned_peer(&mut self, id: PeerId, addresses: Vec<Multiaddr>) -> usize {
349 let mut admitted = Vec::new();
350 for address in addresses {
351 if self.position_of_address(&address).is_none() && self.admit_gossip(&address).is_ok() {
352 admitted.push(address);
353 }
354 }
355 if admitted.is_empty() || self.make_room(Origin::Gossip).is_err() {
356 return 0;
357 }
358 let count = admitted.len();
359 self.entries.push(BookEntry {
360 peer_id: Some(id),
361 addrs: admitted,
362 origin: Origin::Gossip,
363 last_success: None,
364 failures: 0,
365 banned_until: None,
366 });
367 count
368 }
369
370 pub fn succeeded(&mut self, address: &Multiaddr, peer: PeerId, local: PeerId, now: u64) {
375 let Ok((address, _)) = split(address.clone()) else {
376 return;
377 };
378 let Some(index) = self.position_of_address(&address) else {
379 return;
380 };
381 if peer == local {
382 self.entries.remove(index);
383 return;
384 }
385 let learnable = matches!(host(&address), Some(Host::Ip(_)));
386 let known_elsewhere = self
387 .position_of_peer(peer)
388 .is_some_and(|other| other != index);
389 let entry = &mut self.entries[index];
390 entry.last_success = Some(now);
391 entry.failures = 0;
392 entry.banned_until = None;
393 if entry.peer_id.is_none() && learnable && !known_elsewhere {
394 entry.peer_id = Some(peer);
395 }
396 }
397
398 pub fn failed(&mut self, address: &Multiaddr, now: u64) {
401 let Ok((address, _)) = split(address.clone()) else {
402 return;
403 };
404 let Some(index) = self.position_of_address(&address) else {
405 return;
406 };
407 self.fail_entry(index, now);
408 }
409
410 pub fn failed_peer(&mut self, peer: PeerId, now: u64) -> bool {
413 let Some(index) = self.position_of_peer(peer) else {
414 return false;
415 };
416 self.fail_entry(index, now);
417 true
418 }
419
420 fn fail_entry(&mut self, index: usize, now: u64) {
421 let entry = &mut self.entries[index];
422 entry.failures = entry.failures.checked_add(1).unwrap_or(u32::MAX);
423 let cap = match entry.origin {
424 Origin::Seed => Self::SEED_BACKOFF_CAP,
425 Origin::Gossip | Origin::Manual => Self::BACKOFF_CAP,
426 };
427 let delay = 2u64
428 .checked_pow(entry.failures - 1)
429 .and_then(|factor| factor.checked_mul(Self::BACKOFF_SECONDS))
430 .map_or(cap, |delay| delay.min(cap));
431 entry.banned_until = Some(now.checked_add(delay).unwrap_or(u64::MAX));
432 }
433
434 pub fn candidates(
439 &self,
440 now: u64,
441 connected: &[PeerId],
442 omitting: &[String],
443 count: usize,
444 ) -> Vec<Multiaddr> {
445 let eligible = |entry: &&BookEntry| {
446 entry.origin != Origin::Seed
447 && entry.banned_until.is_none_or(|until| until <= now)
448 && entry.peer_id.is_none_or(|peer| {
449 !connected.contains(&peer) && !omitting.contains(&peer.to_string())
450 })
451 };
452 let (mut fresh, mut tried): (Vec<&BookEntry>, Vec<&BookEntry>) = self
453 .entries
454 .iter()
455 .filter(eligible)
456 .partition(|entry| entry.last_success.is_none() && entry.failures == 0);
457 fresh.shuffle(&mut rand::thread_rng());
458 tried.sort_by_key(|entry| (entry.banned_until, entry.failures));
459 let mut fresh = fresh.into_iter();
460 let mut tried = tried.into_iter();
461 let mut picked = Vec::new();
462 while picked.len() < count {
463 let (next_fresh, next_tried) = (fresh.next(), tried.next());
464 if next_fresh.is_none() && next_tried.is_none() {
465 break;
466 }
467 picked.extend(next_fresh.into_iter().chain(next_tried));
468 }
469 picked.truncate(count);
470 picked.into_iter().filter_map(dial_address).collect()
471 }
472
473 pub fn seeds(&self, omitting: &[String], now: Option<u64>) -> Vec<Multiaddr> {
477 let ready = |until: u64, now: u64| until <= now || until - now > Self::SEED_BACKOFF_CAP;
478 let mut seeds: Vec<Multiaddr> = self
479 .entries
480 .iter()
481 .filter(|entry| entry.origin == Origin::Seed)
482 .filter(|entry| {
483 now.is_none_or(|now| entry.banned_until.is_none_or(|until| ready(until, now)))
484 })
485 .filter(|entry| {
486 entry
487 .peer_id
488 .is_none_or(|peer| !omitting.contains(&peer.to_string()))
489 })
490 .filter_map(dial_address)
491 .collect();
492 seeds.shuffle(&mut rand::thread_rng());
493 seeds
494 }
495
496 pub fn sync_addresses(&self, omitting: &[String]) -> Vec<PeerAddress> {
499 self.entries
500 .iter()
501 .filter_map(|entry| {
502 let peer = entry.peer_id?;
503 if omitting.contains(&peer.to_string()) {
504 return None;
505 }
506 entry.addrs.iter().find_map(|address| {
507 PeerAddress::new(peer, address.clone().with(Protocol::P2p(peer))).ok()
508 })
509 })
510 .collect()
511 }
512
513 pub fn gossip(&self, limit: usize, omitting: &[String]) -> Vec<GossipPeer> {
516 let mut verified: Vec<(&BookEntry, PeerId, u64)> = self
517 .entries
518 .iter()
519 .filter_map(|entry| Some((entry, entry.peer_id?, entry.last_success?)))
520 .filter(|(_, peer, _)| !omitting.contains(&peer.to_string()))
521 .collect();
522 verified.sort_by_key(|(_, _, seen)| std::cmp::Reverse(*seen));
523 verified
524 .into_iter()
525 .filter_map(|(entry, peer_id, last_seen)| {
526 let addrs: Vec<Multiaddr> =
527 entry.addrs.iter().filter(|a| public(a)).cloned().collect();
528 (!addrs.is_empty()).then_some(GossipPeer {
529 peer_id,
530 addrs,
531 last_seen,
532 })
533 })
534 .take(limit)
535 .collect()
536 }
537
538 pub fn stats(&self) -> BookStats {
539 let count = |keep: &dyn Fn(&BookEntry) -> bool| {
540 u64::try_from(self.entries.iter().filter(|entry| keep(entry)).count())
541 .unwrap_or(u64::MAX)
542 };
543 BookStats {
544 seeds: count(&|entry| entry.origin == Origin::Seed),
545 book: count(&|_| true),
546 reached: count(&|entry| entry.failures == 0 && entry.last_success.is_some()),
547 reached_ever: count(&|entry| entry.last_success.is_some()),
548 gossip: count(&|entry| entry.origin == Origin::Gossip),
549 }
550 }
551
552 pub fn listing(&self) -> Vec<ListedEntry> {
554 let mut entries: Vec<ListedEntry> = self
555 .entries
556 .iter()
557 .map(|entry| persist(entry).into())
558 .collect();
559 entries.sort_by(|a, b| (&a.peer_id, &a.addrs).cmp(&(&b.peer_id, &b.addrs)));
560 entries.truncate(Self::CAPACITY);
561 entries
562 }
563
564 pub fn remove(&mut self, peer: PeerId) -> bool {
566 let before = self.entries.len();
567 self.entries.retain(|entry| entry.peer_id != Some(peer));
568 self.entries.len() != before
569 }
570
571 fn position_of_address(&self, address: &Multiaddr) -> Option<usize> {
572 self.entries
573 .iter()
574 .position(|entry| entry.addrs.contains(address))
575 }
576
577 fn position_of_peer(&self, peer: PeerId) -> Option<usize> {
578 self.entries
579 .iter()
580 .position(|entry| entry.peer_id == Some(peer))
581 }
582
583 fn overlaps(&self, candidate: &BookEntry) -> bool {
584 candidate
585 .peer_id
586 .is_some_and(|peer| self.position_of_peer(peer).is_some())
587 || candidate
588 .addrs
589 .iter()
590 .any(|address| self.position_of_address(address).is_some())
591 }
592
593 fn admit_gossip(&self, address: &Multiaddr) -> Result<(), Error> {
595 let Some(Host::Ip(ip)) = host(address) else {
596 return Err(Error::Invalid(
597 "gossip addresses must be literal IPs".into(),
598 ));
599 };
600 validate_gossip_address(address)?;
601 if !is_public(address) {
602 return Err(Error::Invalid("gossip named a non-public address".into()));
603 }
604 let neighbours = self
605 .entries
606 .iter()
607 .filter(|entry| entry.origin == Origin::Gossip)
608 .filter(|entry| entry.addrs.iter().any(|known| same_group(known, ip)))
609 .count();
610 if neighbours >= Self::GROUP_LIMIT && self.position_of_address(address).is_none() {
611 return Err(Error::Busy("gossip network group is full".into()));
612 }
613 Ok(())
614 }
615
616 fn make_room(&mut self, origin: Origin) -> Result<(), Error> {
619 if self.entries.len() < Self::CAPACITY {
620 return Ok(());
621 }
622 let worst = self
623 .entries
624 .iter()
625 .enumerate()
626 .filter(|(_, entry)| entry.origin == Origin::Gossip)
627 .max_by_key(|(_, entry)| {
628 (
629 entry.failures,
630 std::cmp::Reverse(entry.last_success.unwrap_or(0)),
631 )
632 })
633 .map(|(index, _)| index);
634 match worst {
635 Some(index)
636 if origin != Origin::Gossip
637 || self.entries[index].failures > 0
638 || self.entries[index].last_success.is_none() =>
639 {
640 self.entries.remove(index);
641 Ok(())
642 }
643 _ => Err(Error::Busy("peer book is full".into())),
644 }
645 }
646}
647
648fn split(address: Multiaddr) -> Result<(Multiaddr, Option<PeerId>), Error> {
651 let invalid = || Error::Invalid("peer book address shape is not dialable".into());
652 let mut bare = Multiaddr::empty();
653 let mut protocols = address.iter();
654 let first = protocols.next().ok_or_else(invalid)?;
655 let named_only = matches!(first, Protocol::Dnsaddr(_));
656 match &first {
657 Protocol::Ip4(ip) if usable(IpAddr::V4(*ip)) => (),
658 Protocol::Ip6(ip) if usable(IpAddr::V6(*ip)) => (),
659 Protocol::Dns(_) | Protocol::Dns4(_) | Protocol::Dns6(_) | Protocol::Dnsaddr(_) => (),
660 _ => return Err(invalid()),
661 }
662 bare.push(first);
663 if !named_only {
664 match protocols.next() {
665 Some(Protocol::Tcp(port)) if port != 0 => bare.push(Protocol::Tcp(port)),
666 _ => return Err(invalid()),
667 }
668 }
669 let peer = match protocols.next() {
670 None => None,
671 Some(Protocol::P2p(peer)) => Some(peer),
672 Some(_) => return Err(invalid()),
673 };
674 if protocols.next().is_some() {
675 return Err(invalid());
676 }
677 Ok((bare, peer))
678}
679
680fn host(address: &Multiaddr) -> Option<Host> {
681 match address.iter().next()? {
682 Protocol::Ip4(ip) => Some(Host::Ip(IpAddr::V4(ip))),
683 Protocol::Ip6(ip) => Some(Host::Ip(IpAddr::V6(ip))),
684 Protocol::Dns(_) | Protocol::Dns4(_) | Protocol::Dns6(_) | Protocol::Dnsaddr(_) => {
685 Some(Host::Name)
686 }
687 _ => None,
688 }
689}
690
691fn usable(ip: IpAddr) -> bool {
692 match ip {
693 IpAddr::V4(ip) => !ip.is_unspecified() && !ip.is_multicast() && !ip.is_broadcast(),
694 IpAddr::V6(ip) => !ip.is_unspecified() && !ip.is_multicast(),
695 }
696}
697
698fn public(address: &Multiaddr) -> bool {
701 matches!(host(address), Some(Host::Ip(_))) && is_public(address)
702}
703
704fn same_group(address: &Multiaddr, ip: IpAddr) -> bool {
706 match (host(address), ip) {
707 (Some(Host::Ip(IpAddr::V4(known))), IpAddr::V4(ip)) => {
708 known.octets()[..3] == ip.octets()[..3]
709 }
710 (Some(Host::Ip(IpAddr::V6(known))), IpAddr::V6(ip)) => {
711 known.segments()[..3] == ip.segments()[..3]
712 }
713 _ => false,
714 }
715}
716
717fn dial_address(entry: &BookEntry) -> Option<Multiaddr> {
718 let address = entry.addrs.first()?.clone();
719 Some(match entry.peer_id {
720 Some(peer) => address.with(Protocol::P2p(peer)),
721 None => address,
722 })
723}
724
725fn persist(entry: &BookEntry) -> SavedEntry {
726 SavedEntry {
727 peer_id: entry.peer_id.map(|peer| peer.to_string()),
728 addrs: entry.addrs.iter().map(Multiaddr::to_string).collect(),
729 origin: entry.origin,
730 last_success: entry.last_success,
731 failures: entry.failures,
732 banned_until: entry.banned_until,
733 }
734}
735
736fn restore(saved: SavedEntry) -> Result<BookEntry, Error> {
737 let peer_id = saved
738 .peer_id
739 .map(|peer| peer.parse::<PeerId>())
740 .transpose()
741 .map_err(|_| Error::Invalid("invalid saved peer ID".into()))?;
742 if saved.addrs.is_empty() || saved.addrs.len() > PeerBook::ADDRESSES {
743 return Err(Error::Invalid(
744 "saved peer needs one to four addresses".into(),
745 ));
746 }
747 let mut addrs = Vec::new();
748 for address in saved.addrs {
749 let parsed: Multiaddr = address
750 .parse()
751 .map_err(|_| Error::Invalid("invalid saved peer address".into()))?;
752 let (bare, named) = split(parsed)?;
753 if named.is_some() || addrs.contains(&bare) {
754 return Err(Error::Invalid("saved peer address is not canonical".into()));
755 }
756 if saved.origin == Origin::Gossip && !public(&bare) {
757 return Err(Error::Invalid("saved gossip address is not public".into()));
758 }
759 addrs.push(bare);
760 }
761 Ok(BookEntry {
762 peer_id,
763 addrs,
764 origin: saved.origin,
765 last_success: saved.last_success,
766 failures: saved.failures,
767 banned_until: saved.banned_until,
768 })
769}
770
771fn io_source(error: urma_io::Error) -> Error {
772 Error::Source(Box::new(error))
773}