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 => match self.network.dns().admits(address) {
199                // Without a resolver the address is unreachable here, not bad.
200                Err(unresolvable) => return Ok(Err(unresolvable)),
201                Ok(()) => self.network.dial(address.clone(), stop, deadline),
202            },
203        };
204        if stop.load(Ordering::Acquire) {
205            return Ok(outcome);
206        }
207        let at = now()?;
208        let mut book = self.lock()?;
209        match &outcome {
210            Ok(peer) => book.succeeded(address, *peer, self.network.peer_id(), at),
211            Err(Error::LocalPeer) if names_self(address, &local) => {
212                self.own()?.push(address.clone())
213            }
214            // A name may resolve to other hosts too: skip only this dial.
215            Err(Error::LocalPeer) => (),
216            // Our own capacity or a busy peer says nothing about the address.
217            Err(Error::DialDenied(_) | Error::Busy(_) | Error::PeerBusy) => (),
218            Err(_failure) => book.failed(address, at),
219        }
220        Ok(outcome)
221    }
222
223    fn own(&self) -> Result<MutexGuard<'_, Vec<Multiaddr>>, Error> {
224        self.own_addresses.lock().map_err(|_| poisoned())
225    }
226
227    /// The startup pass ignores backoff persisted by an earlier process.
228    fn dial_seeds(
229        &self,
230        omitting: &[String],
231        budget: &mut Budget,
232        startup: bool,
233    ) -> Result<bool, Error> {
234        let backoff_at = if startup { None } else { Some(now()?) };
235        let mut seeds = self.lock()?.seeds(omitting, backoff_at);
236        let own = self.own()?.clone();
237        seeds.retain(|seed| !own.contains(seed));
238        let mut dialed = false;
239        let mut error = None;
240        let mut reached = false;
241        for seed in seeds {
242            if !budget.take() {
243                break;
244            }
245            match self.dial(&seed, budget.stop)? {
246                Ok(_peer) => reached = true,
247                Err(Error::LocalPeer) => continue,
248                Err(failure) => error = Some(format!("{seed}: {failure}")),
249            }
250            dialed = true;
251        }
252        if dialed {
253            let recorded = if reached {
254                None
255            } else {
256                error.map(RecordedError::now)
257            };
258            *self.seed_error.lock().map_err(|_| poisoned())? = recorded;
259        }
260        Ok(dialed)
261    }
262
263    fn dial_candidates(&self, omitting: &[String], budget: &mut Budget) -> Result<bool, Error> {
264        let connected = self.network.connected_peer_ids();
265        if connected.len() >= Self::TARGET {
266            return Ok(false);
267        }
268        let mut candidates =
269            self.lock()?
270                .candidates(now()?, &connected, omitting, Self::DIALS_PER_ROUND);
271        let own = self.own()?.clone();
272        candidates.retain(|candidate| !own.contains(candidate));
273        let mut dialed = false;
274        for candidate in candidates {
275            if !budget.take() {
276                break;
277            }
278            dialed = true;
279            // Either outcome is already recorded in the book (success or backoff).
280            match self.dial(&candidate, budget.stop)? {
281                Ok(_reached) => (),
282                Err(_backed_off) => (),
283            }
284        }
285        Ok(dialed)
286    }
287
288    /// Ask one random connected, non-omitting peer for addresses and merge
289    /// them under the book's caps. Lock order: the book lock is released
290    /// before `gossip_error` is taken; the two are never held together.
291    fn ask_for_peers(&self, omitting: &[String], budget: &mut Budget) -> Result<bool, Error> {
292        let connected: Vec<PeerId> = self
293            .network
294            .connected_peer_ids()
295            .into_iter()
296            .filter(|peer| !omitting.contains(&peer.to_string()))
297            .collect();
298        let Some(peer) = connected.choose(&mut rand::thread_rng()).copied() else {
299            return Ok(false);
300        };
301        if !budget.take() {
302            return Ok(false);
303        }
304        let deadline = Instant::now() + Self::OPERATION_TIME;
305        let outcome = self
306            .network
307            .peers_of(peer, Self::GOSSIP_LIMIT, budget.stop, deadline);
308        let (changed, recorded) = match outcome {
309            Ok(peers) => {
310                let local = self.network.peer_id();
311                (self.lock()?.merge(&peers, local, omitting) > 0, None)
312            }
313            Err(failure) => (
314                false,
315                Some(RecordedError::now(format!("{peer}: {failure}"))),
316            ),
317        };
318        *self.gossip_error.lock().map_err(|_| poisoned())? = recorded;
319        Ok(changed)
320    }
321
322    pub fn status(&self) -> Result<DiscoveryStatus, Error> {
323        let stats = self.lock()?.stats();
324        let (last_seed_round_error, last_seed_round_error_at) =
325            RecordedError::split(read(&self.seed_error, "seed round")?);
326        let (last_gossip_error, last_gossip_error_at) =
327            RecordedError::split(read(&self.gossip_error, "gossip")?);
328        let mut status = DiscoveryStatus::default();
329        status.seeds = stats.seeds;
330        status.book = stats.book;
331        status.reached = stats.reached;
332        status.reached_ever = stats.reached_ever;
333        status.gossip = stats.gossip;
334        status.last_seed_round_error = last_seed_round_error;
335        status.last_seed_round_error_at = last_seed_round_error_at;
336        status.last_gossip_error = last_gossip_error;
337        status.last_gossip_error_at = last_gossip_error_at;
338        Ok(status)
339    }
340
341    /// Peers the inventory scheduler caught misbehaving back off in the book
342    /// too, so discovery stops dialing them; the book is saved if it changed.
343    pub fn penalize(&self, peers: &[PeerId]) -> Result<(), Error> {
344        if peers.is_empty() {
345            return Ok(());
346        }
347        let at = now()?;
348        let mut book = self.lock()?;
349        let mut changed = false;
350        for peer in peers {
351            changed |= book.failed_peer(*peer, at);
352        }
353        if changed {
354            book.save(&self.path)?;
355        }
356        Ok(())
357    }
358
359    pub fn book(&self) -> Result<Vec<BookEntry>, Error> {
360        Ok(self.lock()?.listing())
361    }
362}
urma_node/peer_discovery.rs

urma_node/
peer_discovery.rs
