urma_node/
peer_discovery.rs

1//! Bitcoin-style peer discovery. Seeds bootstrap a sparse book; afterwards
2//! connected peers exchange verified addresses. A dial proves only that a
3//! Noise identity on this chain answered at an address: never record
4//! availability or honesty. Each Service round performs at most three
5//! network operations and checks cancellation between them.
6use crate::{
7    Error,
8    engine::Engine,
9    model::{BookEntry, DiscoveryStatus},
10    peer::PeerNetwork,
11    peer_protocol::GossipPeer,
12    peerbook::PeerBook,
13    recorded::{RecordedError, read, unix_seconds as now},
14};
15use libp2p::{Multiaddr, PeerId, multiaddr::Protocol};
16use rand::seq::SliceRandom;
17use std::{
18    path::PathBuf,
19    sync::{
20        Arc, Mutex, MutexGuard,
21        atomic::{AtomicBool, AtomicU64, Ordering},
22    },
23    time::{Duration, Instant},
24};
25
26pub struct Discovery {
27    network: Arc<PeerNetwork>,
28    book: Arc<Mutex<PeerBook>>,
29    path: PathBuf,
30    round: AtomicU64,
31    seed_error: Mutex<Option<RecordedError>>,
32    gossip_error: Mutex<Option<RecordedError>>,
33    sync_backoff: Mutex<Vec<String>>,
34    own_addresses: Mutex<Vec<Multiaddr>>,
35}
36
37/// Network operations left in this round; stop ends the round early.
38struct Budget<'a> {
39    left: usize,
40    stop: &'a AtomicBool,
41}
42
43impl Budget<'_> {
44    fn take(&mut self) -> bool {
45        if self.left == 0 || self.stop.load(Ordering::Acquire) {
46            return false;
47        }
48        self.left -= 1;
49        true
50    }
51}
52
53/// An address that can only ever be this node: it names our identity, or is
54/// an IP literal (no name that might also resolve elsewhere).
55fn names_self(address: &Multiaddr, local: &Protocol) -> bool {
56    let literal = matches!(
57        address.iter().next(),
58        Some(Protocol::Ip4(_) | Protocol::Ip6(_))
59    );
60    literal || address.iter().any(|part| part == *local)
61}
62
63fn poisoned() -> Error {
64    Error::Invalid("peer discovery state poisoned".into())
65}
66
67impl Discovery {
68    /// Connected peers the round tries to reach.
69    pub const TARGET: usize = 8;
70    pub const DIALS_PER_ROUND: usize = 2;
71    pub const OPERATIONS_PER_ROUND: usize = 3;
72    /// Rounds between address requests to one random connected peer.
73    pub const GOSSIP_EVERY: u64 = 30;
74    pub const GOSSIP_LIMIT: u32 = 32;
75    /// Rounds between seed passes while below target but connected.
76    pub const SEED_EVERY: u64 = 600;
77    /// At startup seeds are dialed while fewer entries than this were ever reached.
78    pub const BOOTSTRAP_REACHED: u64 = 4;
79    /// Wall-clock bound of one dial or address request.
80    pub const OPERATION_TIME: Duration = Duration::from_secs(10);
81
82    /// `path` is where the book is saved after a round changes it.
83    pub fn new(network: Arc<PeerNetwork>, book: Arc<Mutex<PeerBook>>, path: PathBuf) -> Self {
84        Self {
85            network,
86            book,
87            path,
88            round: AtomicU64::new(0),
89            seed_error: Mutex::new(None),
90            gossip_error: Mutex::new(None),
91            sync_backoff: Mutex::new(Vec::new()),
92            own_addresses: Mutex::new(Vec::new()),
93        }
94    }
95
96    /// Peers inventory sync is holding off (evicted, idle or blamed): neither
97    /// seed passes nor candidate dials reconnect them until they leave this set.
98    pub fn hold_off(&self, peers: Vec<String>) -> Result<(), Error> {
99        *self.sync_backoff.lock().map_err(|_| poisoned())? = peers;
100        Ok(())
101    }
102
103    fn lock(&self) -> Result<MutexGuard<'_, PeerBook>, Error> {
104        self.book.lock().map_err(|_| poisoned())
105    }
106
107    /// One bounded round: merge peers learned through identify, refresh the
108    /// served gossip snapshot, seeds when due, up to two dials while below
109    /// target, every thirtieth round one address request.
110    pub fn step(&self, engine: &Engine, stop: &AtomicBool) -> Result<(), Error> {
111        let round = self.round.fetch_add(1, Ordering::Relaxed);
112        let omitting = engine.omitting_peers()?;
113        let mut changed = self.merge_identified(&omitting)?;
114        self.publish(&omitting)?;
115        let mut budget = Budget {
116            left: Self::OPERATIONS_PER_ROUND,
117            stop,
118        };
119        let mut avoid = omitting.clone();
120        avoid.extend(
121            self.sync_backoff
122                .lock()
123                .map_err(|_| poisoned())?
124                .iter()
125                .cloned(),
126        );
127        if self.seeds_due(round)? {
128            changed |= self.dial_seeds(&avoid, &mut budget, round == 0)?;
129        }
130        changed |= self.dial_candidates(&avoid, &mut budget)?;
131        if round % Self::GOSSIP_EVERY == Self::GOSSIP_EVERY - 1 {
132            changed |= self.ask_for_peers(&omitting, &mut budget)?;
133        }
134        if changed {
135            self.lock()?.save(&self.path)?;
136        }
137        Ok(())
138    }
139
140    /// Peers that identified themselves to us, with public literal addresses,
141    /// enter the book as gossip under the same caps; a local operation.
142    fn merge_identified(&self, omitting: &[String]) -> Result<bool, Error> {
143        let at = now()?;
144        let peers: Vec<GossipPeer> = self
145            .network
146            .drain_identified()
147            .into_iter()
148            .map(|(peer_id, addrs)| GossipPeer {
149                peer_id,
150                addrs,
151                last_seen: at,
152            })
153            .collect();
154        if peers.is_empty() {
155            return Ok(false);
156        }
157        let local = self.network.peer_id();
158        Ok(self.lock()?.merge_identified(&peers, local, omitting) > 0)
159    }
160
161    /// Verified book entries; this node itself is learned through identify.
162    fn publish(&self, omitting: &[String]) -> Result<(), Error> {
163        let limit = usize::try_from(Self::GOSSIP_LIMIT)
164            .map_err(|error| Error::Invalid(format!("gossip limit: {error}")))?;
165        let at = now()?;
166        let mut peers = Vec::new();
167        for mut peer in self.lock()?.gossip(limit, omitting) {
168            peer.last_seen = peer.last_seen.min(at);
169            peers.push(peer);
170        }
171        self.network.set_gossip(peers);
172        Ok(())
173    }
174
175    /// Isolated nodes retry seeds every round, limited by each seed's backoff.
176    fn seeds_due(&self, round: u64) -> Result<bool, Error> {
177        let connected = self.network.connected_peer_ids().len();
178        if connected >= Self::TARGET {
179            return Ok(false);
180        }
181        if connected == 0 {
182            return Ok(true);
183        }
184        let reached_ever = self.lock()?.stats().reached_ever;
185        Ok((round == 0 && reached_ever < Self::BOOTSTRAP_REACHED)
186            || (round > 0 && round % Self::SEED_EVERY == 0))
187    }
188
189    /// Dial one address outside the book lock and record the outcome. An
190    /// address that reached this node itself (a seed resolving to us) is
191    /// remembered and never dialed again by this process; the book keeps it,
192    /// since the same name is a real peer for every other node.
193    fn dial(&self, address: &Multiaddr, stop: &AtomicBool) -> Result<Result<PeerId, Error>, Error> {
194        let deadline = Instant::now() + Self::OPERATION_TIME;
195        let local = Protocol::P2p(self.network.peer_id());
196        let outcome = match address.iter().any(|part| part == local) {
197            true => Err(Error::LocalPeer),
198            false => self.network.dial(address.clone(), stop, deadline),
199        };
200        if stop.load(Ordering::Acquire) {
201            return Ok(outcome);
202        }
203        let at = now()?;
204        let mut book = self.lock()?;
205        match &outcome {
206            Ok(peer) => book.succeeded(address, *peer, self.network.peer_id(), at),
207            Err(Error::LocalPeer) if names_self(address, &local) => {
208                self.own()?.push(address.clone())
209            }
210            // A name may resolve to other hosts too: skip only this dial.
211            Err(Error::LocalPeer) => (),
212            // Our own capacity or a busy peer says nothing about the address.
213            Err(Error::DialDenied(_) | Error::Busy(_) | Error::PeerBusy) => (),
214            Err(_failure) => book.failed(address, at),
215        }
216        Ok(outcome)
217    }
218
219    fn own(&self) -> Result<MutexGuard<'_, Vec<Multiaddr>>, Error> {
220        self.own_addresses.lock().map_err(|_| poisoned())
221    }
222
223    /// The startup pass ignores backoff persisted by an earlier process.
224    fn dial_seeds(
225        &self,
226        omitting: &[String],
227        budget: &mut Budget,
228        startup: bool,
229    ) -> Result<bool, Error> {
230        let backoff_at = if startup { None } else { Some(now()?) };
231        let mut seeds = self.lock()?.seeds(omitting, backoff_at);
232        let own = self.own()?.clone();
233        seeds.retain(|seed| !own.contains(seed));
234        let mut dialed = false;
235        let mut error = None;
236        let mut reached = false;
237        for seed in seeds {
238            if !budget.take() {
239                break;
240            }
241            match self.dial(&seed, budget.stop)? {
242                Ok(_peer) => reached = true,
243                Err(Error::LocalPeer) => continue,
244                Err(failure) => error = Some(format!("{seed}: {failure}")),
245            }
246            dialed = true;
247        }
248        if dialed {
249            let recorded = if reached {
250                None
251            } else {
252                error.map(RecordedError::now)
253            };
254            *self.seed_error.lock().map_err(|_| poisoned())? = recorded;
255        }
256        Ok(dialed)
257    }
258
259    fn dial_candidates(&self, omitting: &[String], budget: &mut Budget) -> Result<bool, Error> {
260        let connected = self.network.connected_peer_ids();
261        if connected.len() >= Self::TARGET {
262            return Ok(false);
263        }
264        let mut candidates =
265            self.lock()?
266                .candidates(now()?, &connected, omitting, Self::DIALS_PER_ROUND);
267        let own = self.own()?.clone();
268        candidates.retain(|candidate| !own.contains(candidate));
269        let mut dialed = false;
270        for candidate in candidates {
271            if !budget.take() {
272                break;
273            }
274            dialed = true;
275            // Either outcome is already recorded in the book (success or backoff).
276            match self.dial(&candidate, budget.stop)? {
277                Ok(_reached) => (),
278                Err(_backed_off) => (),
279            }
280        }
281        Ok(dialed)
282    }
283
284    /// Ask one random connected, non-omitting peer for addresses and merge
285    /// them under the book's caps. Lock order: the book lock is released
286    /// before `gossip_error` is taken; the two are never held together.
287    fn ask_for_peers(&self, omitting: &[String], budget: &mut Budget) -> Result<bool, Error> {
288        let connected: Vec<PeerId> = self
289            .network
290            .connected_peer_ids()
291            .into_iter()
292            .filter(|peer| !omitting.contains(&peer.to_string()))
293            .collect();
294        let Some(peer) = connected.choose(&mut rand::thread_rng()).copied() else {
295            return Ok(false);
296        };
297        if !budget.take() {
298            return Ok(false);
299        }
300        let deadline = Instant::now() + Self::OPERATION_TIME;
301        let outcome = self
302            .network
303            .peers_of(peer, Self::GOSSIP_LIMIT, budget.stop, deadline);
304        let (changed, recorded) = match outcome {
305            Ok(peers) => {
306                let local = self.network.peer_id();
307                (self.lock()?.merge(&peers, local, omitting) > 0, None)
308            }
309            Err(failure) => (
310                false,
311                Some(RecordedError::now(format!("{peer}: {failure}"))),
312            ),
313        };
314        *self.gossip_error.lock().map_err(|_| poisoned())? = recorded;
315        Ok(changed)
316    }
317
318    pub fn status(&self) -> Result<DiscoveryStatus, Error> {
319        let stats = self.lock()?.stats();
320        let (last_seed_round_error, last_seed_round_error_at) =
321            RecordedError::split(read(&self.seed_error, "seed round")?);
322        let (last_gossip_error, last_gossip_error_at) =
323            RecordedError::split(read(&self.gossip_error, "gossip")?);
324        let mut status = DiscoveryStatus::default();
325        status.seeds = stats.seeds;
326        status.book = stats.book;
327        status.reached = stats.reached;
328        status.reached_ever = stats.reached_ever;
329        status.gossip = stats.gossip;
330        status.last_seed_round_error = last_seed_round_error;
331        status.last_seed_round_error_at = last_seed_round_error_at;
332        status.last_gossip_error = last_gossip_error;
333        status.last_gossip_error_at = last_gossip_error_at;
334        Ok(status)
335    }
336
337    /// Peers the inventory scheduler caught misbehaving back off in the book
338    /// too, so discovery stops dialing them; the book is saved if it changed.
339    pub fn penalize(&self, peers: &[PeerId]) -> Result<(), Error> {
340        if peers.is_empty() {
341            return Ok(());
342        }
343        let at = now()?;
344        let mut book = self.lock()?;
345        let mut changed = false;
346        for peer in peers {
347            changed |= book.failed_peer(*peer, at);
348        }
349        if changed {
350            book.save(&self.path)?;
351        }
352        Ok(())
353    }
354
355    pub fn book(&self) -> Result<Vec<BookEntry>, Error> {
356        Ok(self.lock()?.listing())
357    }
358}
urma_node/peer_discovery.rs

urma_node/
peer_discovery.rs
