diff options
Diffstat (limited to 'src/transport/paths.rs')
| -rw-r--r-- | src/transport/paths.rs | 1247 |
1 files changed, 1247 insertions, 0 deletions
diff --git a/src/transport/paths.rs b/src/transport/paths.rs new file mode 100644 index 0000000..ba16794 --- /dev/null +++ b/src/transport/paths.rs @@ -0,0 +1,1247 @@ +//! Paths: one address space over any number of transports, below the identity layer +//! (`transport::identity`). An address is `<scheme>:<target>`: the transport by its +//! scheme (`mycelium`, `libp2p`, …), and on it a transport identity (Mycelium's public +//! key, libp2p's peer ID) or a location (`Transport::locate`), e.g. a multiaddr. +//! Addresses as paths keeps them are `<scheme>:<transport identity>` (`canonical`). +//! +//! Every transport carries the same messages (`wire`); paths routes each send to the +//! transport of its address, hands every transport's messages to one handler under the +//! sender's address, and puts together what the transports say about themselves +//! (addresses, seeds they found, figures). The transports are in order of preference, +//! as configured (`--transports`). +//! +//! A message to a node with several addresses goes to the first of them that takes it +//! (`send_first`), in the order `rank` gives. A send that fails is tried at once at the +//! next address, the message encoded for its transport: when one transport fails, the +//! next carries on within the same send. +//! +//! Paths measures the addresses of such nodes itself (and every path the node's +//! keepalive PINGs go over, below): every `PROBE_EVERY`, each address it had a choice of +//! within `ACTIVE_FOR` gets a PING with a random nonce, which the +//! other side's paths answers at once with a PONG, over the same path. Answers give each +//! path's round-trip time (smoothed); `UNANSWERED_DOWN` PINGs in a row unanswered, or a +//! send that failed within `DOWN_FOR`, make a path down. `rank` puts paths that are up +//! first, the one in use first among them, and the rest by round-trip time; it changes +//! the one in use when another is faster by `SWITCH_GAIN` (a quarter) and at least +//! `SWITCH_MIN` (both measured), so it doesn't flap. Until both are measured, the +//! configured order decides. +//! +//! PINGs and PONGs then go up as other messages do: they tell the node the peer is +//! alive. The node's keepalive is a PING too (`swarm::peer::keepalive`), which paths +//! sends like any message, and notes as the probe of the path it went over. +//! +//! The debug firewall (`transport::firewall`) works here, below everything else, through +//! a `Gate`: messages from an address it blocks are dropped as they arrive, and sends to +//! one are dropped silently, as by a lossy network. + +use std::collections::HashMap; +use std::sync::{Arc, Mutex, Weak}; +use std::time::Duration; + +use anyhow::{Context, Result, anyhow}; +use tokio::time::Instant; +use tracing::{debug, info}; + +use crate::locks::MutexExt; +use crate::transport::stats::{PathCounts, PathInfo, PathsStats, TransportStats}; +use crate::transport::traffic::Traffic; +use crate::transport::{Connection, EncodedMessage, ParsedMessage, Transport}; + +/// Whether to drop what comes from or goes to an address (the debug firewall, +/// `transport::firewall`; the identity layer knows which node is where). +pub(crate) trait Gate: Send + Sync { + fn blocks_inbound(&self, address: &str) -> bool; + fn blocks_outbound(&self, address: &str) -> bool; +} + +/// Called with the sender's address, the message and its `payload_len`. +pub(crate) type PathHandler = dyn Fn(String, ParsedMessage, usize) -> std::pin::Pin<Box<dyn Future<Output = Result<()>> + Send>> + + Send + + Sync + + 'static; + +/// One of the node's transports. +struct Link { + scheme: &'static str, + inner: Arc<dyn Transport>, +} + +impl Link { + fn address(&self, target: &str) -> String { + format!("{}:{target}", self.scheme) + } + + fn own_address(&self) -> String { + self.address(self.inner.local_id()) + } +} + +/// The scheme of `address`. +pub(crate) fn scheme_of(address: &str) -> &str { + address.split_once(':').map_or("", |(scheme, _)| scheme) +} + +/// A path whose send failed is down for this long (unless it answers a PING, or a +/// send there works again). +const DOWN_FOR: Duration = Duration::from_secs(60); +/// How often a path in use is pinged, how long an answer may take, and how many PINGs +/// in a row unanswered make it down. +const PROBE_EVERY: Duration = Duration::from_secs(30); +const PROBE_TIMEOUT: Duration = Duration::from_secs(15); +const UNANSWERED_DOWN: u32 = 3; +/// How often the probe task looks for paths due. +const PROBE_TICK: Duration = Duration::from_secs(5); +/// Paths this node had a choice of within this long are probed; ones it didn't use for +/// twice as long are forgotten. +const ACTIVE_FOR: Duration = Duration::from_secs(300); +/// The path in use is changed for one this much faster (a fraction of its round-trip +/// time), and at least `SWITCH_MIN`. +const SWITCH_GAIN: f64 = 0.25; +const SWITCH_MIN: Duration = Duration::from_millis(5); +/// Paths kept at most: they are of peers this node chose among, but bounded anyway. +const MAX_PATHS: usize = 4096; + +/// What paths knows about one path (an address as paths keep it). +#[derive(Debug, Default)] +struct PathState { + /// When it was last among several to choose from (`rank`). + used: Option<Instant>, + /// Smoothed round-trip time of its PINGs. + rtt: Option<Duration>, + /// The PING out, by nonce, and when it went. + probe: Option<(u64, Instant)>, + last_probe: Option<Instant>, + /// PINGs in a row unanswered. + unanswered: u32, + /// When a send there last failed. + failed_at: Option<Instant>, +} + +impl PathState { + fn is_down(&self, now: Instant) -> bool { + self.unanswered >= UNANSWERED_DOWN + || self + .failed_at + .is_some_and(|at| now.saturating_duration_since(at) < DOWN_FOR) + } +} + +/// Every path's state, and which one is in use for each set to choose from. +#[derive(Debug, Default)] +struct Table { + paths: HashMap<String, PathState>, + /// By the set's addresses (sorted, joined): the one in use. + chosen: HashMap<String, String>, + counts: PathCounts, +} + +impl Table { + /// `usable`, with each one's place in the configured order, best first (as + /// `Paths::rank` says), noting the choice. + fn rank(&mut self, mut usable: Vec<(String, usize)>, now: Instant) -> Vec<String> { + usable.sort_by_key(|(_, preference)| *preference); + usable.dedup_by(|a, b| a.0 == b.0); + if usable.len() < 2 { + return usable.into_iter().map(|(address, _)| address).collect(); + } + if self.paths.len() >= MAX_PATHS { + self.forget_unused(now); + } + for (address, _) in &usable { + if self.paths.len() < MAX_PATHS || self.paths.contains_key(address) { + self.paths.entry(address.clone()).or_default().used = Some(now); + } + } + let rtt = |table: &Self, address: &str| table.paths.get(address).and_then(|path| path.rtt); + let (up, down): (Vec<_>, Vec<_>) = usable.into_iter().partition(|(address, _)| { + !self + .paths + .get(address) + .is_some_and(|path| path.is_down(now)) + }); + let mut set: Vec<&str> = up + .iter() + .chain(down.iter()) + .map(|(address, _)| address.as_str()) + .collect(); + set.sort_unstable(); + let key = set.join(" "); + let mut ranked: Vec<String> = Vec::new(); + if !up.is_empty() { + let mut current = self + .chosen + .get(&key) + .filter(|chosen| up.iter().any(|(address, _)| address == *chosen)) + .cloned() + .unwrap_or_else(|| up[0].0.clone()); + let fastest = up + .iter() + .filter_map(|(address, _)| Some((address, rtt(self, address)?))) + .min_by_key(|(_, rtt)| *rtt); + if let (Some((fastest, fastest_rtt)), Some(current_rtt)) = + (fastest, rtt(self, ¤t)) + && *fastest != current + && current_rtt.saturating_sub(fastest_rtt) + >= SWITCH_MIN.max(current_rtt.mul_f64(SWITCH_GAIN)) + { + info!( + from = %current, + to = %fastest, + from_rtt = ?current_rtt, + to_rtt = ?fastest_rtt, + "Switching to a faster path" + ); + self.counts.switches += 1; + current = fastest.clone(); + } + if self.chosen.len() < MAX_PATHS || self.chosen.contains_key(&key) { + self.chosen.insert(key, current.clone()); + } + // The rest: measured ones by round-trip time, then unmeasured ones in the + // configured order. + let mut rest: Vec<&(String, usize)> = up + .iter() + .filter(|(address, _)| *address != current) + .collect(); + rest.sort_by_key(|(address, preference)| { + ( + rtt(self, address).is_none(), + rtt(self, address), + *preference, + ) + }); + ranked.push(current); + ranked.extend(rest.into_iter().map(|(address, _)| address.clone())); + } + ranked.extend(down.into_iter().map(|(address, _)| address)); + ranked + } + + /// A send to `address` failed. + fn failed(&mut self, address: &str, now: Instant) { + self.counts.send_failures += 1; + if let Some(path) = self.paths.get_mut(address) { + path.failed_at = Some(now); + // A PING that didn't go out gets no answer; it is no unanswered one. + path.probe = None; + } + } + + /// A send to `address` worked. + fn sent(&mut self, address: &str) { + if let Some(path) = self.paths.get_mut(address) { + path.failed_at = None; + } + } + + /// The PINGs due now, with their nonces (from `nonce`): to every path used within + /// `ACTIVE_FOR` that wasn't pinged within `PROBE_EVERY` and has no PING out. A PING + /// out for `PROBE_TIMEOUT` counts as unanswered. + fn due(&mut self, now: Instant, mut nonce: impl FnMut() -> u64) -> Vec<(String, u64)> { + self.forget_unused(now); + let mut due = Vec::new(); + for (address, path) in &mut self.paths { + if let Some((_, sent)) = path.probe + && now.saturating_duration_since(sent) >= PROBE_TIMEOUT + { + path.probe = None; + path.unanswered += 1; + self.counts.probes_unanswered += 1; + if path.unanswered == UNANSWERED_DOWN { + info!(address = %address, "A path stopped answering PINGs; using others"); + } + } + let active = path + .used + .is_some_and(|used| now.saturating_duration_since(used) < ACTIVE_FOR); + let rested = path + .last_probe + .is_none_or(|last| now.saturating_duration_since(last) >= PROBE_EVERY); + if active && rested && path.probe.is_none() { + let nonce = nonce(); + path.probe = Some((nonce, now)); + path.last_probe = Some(now); + self.counts.probes_sent += 1; + due.push((address.clone(), nonce)); + } + } + due + } + + /// The node sent a PING with `nonce` to `address` (its keepalive): the probe of that + /// path, which its PONG measures, even if it is a peer's only path (then it isn't + /// pinged otherwise: `due` pings only paths there was a choice of). + fn pinged(&mut self, address: &str, nonce: u64, now: Instant) { + if self.paths.len() >= MAX_PATHS && !self.paths.contains_key(address) { + self.forget_unused(now); + if self.paths.len() >= MAX_PATHS { + return; + } + } + let path = self.paths.entry(address.to_owned()).or_default(); + path.probe = Some((nonce, now)); + path.last_probe = Some(now); + self.counts.probes_sent += 1; + } + + /// The PING with `nonce` to `address` (from `due`) goes now: its round trip counts + /// from here, not from when it was found due. + fn sending(&mut self, address: &str, nonce: u64, now: Instant) { + if let Some(path) = self.paths.get_mut(address) + && let Some((expected, sent)) = &mut path.probe + && *expected == nonce + { + *sent = now; + } + } + + /// A PONG with `nonce` from `address`: the round trip, if it answers our PING out. + fn pong(&mut self, address: &str, nonce: u64, now: Instant) -> Option<Duration> { + let path = self.paths.get_mut(address)?; + let (expected, sent) = path.probe?; + if expected != nonce { + return None; + } + let sample = now.saturating_duration_since(sent); + path.probe = None; + if path.unanswered >= UNANSWERED_DOWN { + info!(address = %address, "A path answers PINGs again"); + } + path.unanswered = 0; + path.failed_at = None; + path.rtt = Some(match path.rtt { + Some(rtt) => (rtt * 3 + sample) / 4, + None => sample, + }); + self.counts.probes_answered += 1; + path.rtt + } + + /// Forgets the paths neither used nor pinged for twice `ACTIVE_FOR`, and choices + /// among them. + fn forget_unused(&mut self, now: Instant) { + let recent = |at: Option<Instant>| { + at.is_some_and(|at| now.saturating_duration_since(at) < ACTIVE_FOR * 2) + }; + self.paths + .retain(|_, path| recent(path.used) || recent(path.last_probe)); + let paths = &self.paths; + self.chosen + .retain(|key, _| key.split(' ').all(|address| paths.contains_key(address))); + } + + fn info(&self, address: &str, now: Instant) -> PathInfo { + let path = self.paths.get(address); + PathInfo { + address: address.to_owned(), + rtt_ms: path + .and_then(|path| path.rtt) + .map(|rtt| rtt.as_secs_f64() * 1000.0), + down: path.is_some_and(|path| path.is_down(now)), + } + } +} + +pub(crate) struct Paths { + /// In order of preference. + links: Vec<Link>, + /// Its own PINGs and PONGs are counted here, under the address (`traffic`). + traffic: Option<Arc<Traffic>>, + /// The debug firewall, if any. + gate: Option<Weak<dyn Gate>>, + /// A std mutex: never held across an await. + table: Mutex<Table>, +} + +impl Paths { + /// Over `transports`, by scheme, in order of preference; its own PINGs and PONGs + /// counted in `traffic`, the firewall through `gate`. + pub(crate) fn new( + transports: Vec<(&'static str, Arc<dyn Transport>)>, + traffic: Option<Arc<Traffic>>, + gate: Option<Weak<dyn Gate>>, + ) -> Self { + Self { + links: transports + .into_iter() + .map(|(scheme, inner)| Link { scheme, inner }) + .collect(), + traffic, + gate, + table: Mutex::default(), + } + } + + fn gate(&self) -> Option<Arc<dyn Gate>> { + self.gate.as_ref()?.upgrade() + } + + /// The transport `address` is on, and the transport identity at it + /// (`Transport::locate`, which notes where that is if the address says); None if no + /// transport of ours has its scheme, or it isn't one of that transport's. + fn split(&self, address: &str) -> Option<(&Link, String)> { + let (scheme, target) = address.split_once(':')?; + let link = self.links.iter().find(|link| link.scheme == scheme)?; + let target = link.inner.locate(target)?; + Some((link, target)) + } + + /// Whether one of our transports can use `address`. + pub(crate) fn usable(&self, address: &str) -> bool { + self.split(address).is_some() + } + + /// `address` as paths keep it: `<scheme>:<transport identity>`. + pub(crate) fn canonical(&self, address: &str) -> Option<String> { + self.split(address) + .map(|(link, target)| link.address(&target)) + } + + /// Where the node at `address` (as paths keep it) is, as others can use it + /// (`Transport::locations`). + pub(crate) fn locations(&self, address: &str) -> Vec<String> { + let Some((link, target)) = self.split(address) else { + return Vec::new(); + }; + link.inner + .locations(&target) + .iter() + .map(|location| link.address(location)) + .collect() + } + + /// Where `address`'s transport is in the order of preference (none of ours: last). + pub(crate) fn preference(&self, address: &str) -> usize { + let scheme = scheme_of(address); + self.links + .iter() + .position(|link| link.scheme == scheme) + .unwrap_or(usize::MAX) + } + + /// The usable ones of `addresses` (as paths keep them), best first: paths that are + /// up before those down, the one in use first, then by round-trip time, then in + /// the configured order (module docs). Several addresses are paths to measure. + pub(crate) fn rank(&self, addresses: &[String]) -> Vec<String> { + let usable: Vec<(String, usize)> = addresses + .iter() + .filter(|address| self.usable(address)) + .map(|address| (address.clone(), self.preference(address))) + .collect(); + self.table.lock_or_recover().rank(usable, Instant::now()) + } + + /// The smoothed round-trip time of the path to `address`, if measured. + pub(crate) fn rtt(&self, address: &str) -> Option<Duration> { + self.table.lock_or_recover().paths.get(address)?.rtt + } + + /// What paths knows about each of `addresses`. + pub(crate) fn info(&self, addresses: &[String]) -> Vec<PathInfo> { + let now = Instant::now(); + let infos = { + let table = self.table.lock_or_recover(); + addresses + .iter() + .map(|address| table.info(address, now)) + .collect() + }; + self.located(infos) + } + + /// `infos` with where each path goes, where the transport knows + /// (`Transport::location`); asked outside the table's lock. + fn located(&self, mut infos: Vec<PathInfo>) -> Vec<PathInfo> { + for info in &mut infos { + if let Some((link, location)) = self + .split(&info.address) + .and_then(|(link, target)| Some((link, link.inner.location(&target)?))) + { + info.address = link.address(&location); + } + } + infos + } + + /// Whether `address` is one of ours. + pub(crate) fn is_own(&self, address: &str) -> bool { + self.links.iter().any(|link| link.own_address() == address) + } + + /// Our address on the transport of `address`, as `address` names it there. + pub(crate) fn own_address(&self, address: &str) -> Option<String> { + let scheme = scheme_of(address); + self.links + .iter() + .find(|link| link.scheme == scheme) + .map(Link::own_address) + } + + /// The longest of the transports': until then, something may still be on its way. + pub(crate) fn delivery_limit(&self) -> Duration { + self.links + .iter() + .map(|link| link.inner.delivery_limit()) + .max() + .unwrap_or_default() + } + + /// Each transport's (`Transport::local_addresses`), with its scheme. + pub(crate) fn local_addresses(&self) -> Vec<String> { + self.links + .iter() + .flat_map(|link| { + link.inner + .local_addresses() + .into_iter() + .map(|target| link.address(&target)) + }) + .collect() + } + + /// The schemes of the transports, in order of preference. + pub(crate) fn transports(&self) -> Vec<String> { + self.links + .iter() + .map(|link| link.scheme.to_owned()) + .collect() + } + + /// Each transport's, with its scheme. + pub(crate) fn discovered_seeds(&self) -> Vec<String> { + self.links + .iter() + .flat_map(|link| { + link.inner + .discovered_seeds() + .into_iter() + .map(|seed| link.address(&seed)) + }) + .collect() + } + + /// The address of the node a seed leads to, as paths keep it: an address of ours + /// (`<scheme>:...`) whose transport reads it at once; with a scheme of ours, what + /// that transport resolves the rest to (an address the transport can't read as an + /// identity, e.g. a libp2p multiaddr without its peer ID, still resolves); without + /// one, what the first transport that can resolves it to. + pub(crate) async fn resolve(&self, seed: &str) -> Result<String> { + if let Some(address) = self.canonical(seed) { + return Ok(address); + } + let scheme_and_rest = seed.split_once(':').and_then(|(scheme, rest)| { + let link = self.links.iter().find(|link| link.scheme == scheme)?; + Some((link, rest)) + }); + if let Some((link, rest)) = scheme_and_rest { + let target = link + .inner + .resolve_seed(rest) + .await + .with_context(|| format!("{}: {seed}", link.scheme))?; + return Ok(link.address(&target)); + } + let mut failures = Vec::new(); + for link in &self.links { + match link.inner.resolve_seed(seed).await { + Ok(target) => return Ok(link.address(&target)), + Err(error) => failures.push(format!("{}: {error:#}", link.scheme)), + } + } + Err(anyhow!("{}", failures.join("; "))) + } + + /// `message` encoded for `address`: `peer` and `address` of the result are + /// `address`, and it keeps the message (to encode it for another address). + pub(crate) fn encode(&self, address: &str, message: &ParsedMessage) -> Result<EncodedMessage> { + let (link, target) = self + .split(address) + .with_context(|| format!("no transport for {address}"))?; + let mut encoded = link.inner.encode(&target, message)?; + encoded.peer = address.to_owned(); + encoded.address = address.to_owned(); + encoded.message = Some(Arc::new(message.clone())); + Ok(encoded) + } + + /// Sends `message` to the first of `addresses` (`rank`ed) that takes it: to its own + /// `address` as encoded, to others encoded again for their transport (if it keeps + /// the message). Returns the address it went to; an error with every address's if + /// none took it. + pub(crate) async fn send_first( + &self, + message: &EncodedMessage, + addresses: &[String], + ) -> Result<String> { + let mut failures: Vec<(String, anyhow::Error)> = Vec::new(); + for address in addresses { + let encoded = if *address == message.address { + message.clone() + } else { + let Some(parsed) = &message.message else { + continue; + }; + match self.encode(address, parsed) { + Ok(mut encoded) => { + encoded.peer = message.peer.clone(); + encoded + } + Err(error) => { + failures.push((address.clone(), error)); + continue; + } + } + }; + // A PING is noted before it goes: on a fast link, its PONG may be back + // before the send returns. + if let Some(ParsedMessage::Ping(nonce)) = message.message.as_deref() { + self.table + .lock_or_recover() + .pinged(address, *nonce, Instant::now()); + } + match self.send(&encoded).await { + Ok(()) => { + let mut table = self.table.lock_or_recover(); + table.sent(address); + if !failures.is_empty() { + table.counts.failovers += 1; + drop(table); + debug!( + peer = %message.peer, + address = %address, + failed = %describe(&failures), + "Sent at another address after a send failed" + ); + } + return Ok(address.clone()); + } + Err(error) => { + self.table.lock_or_recover().failed(address, Instant::now()); + failures.push((address.clone(), error)); + } + } + } + match failures.len() { + 0 => Err(anyhow!("no address to send to")), + // The transport's error as it is. + 1 => Err(failures.pop().expect("one").1), + _ => Err(anyhow!("{}", describe(&failures))), + } + } + + /// Sends `message` to its `address` (dropped silently if the firewall blocks it). + pub(crate) async fn send(&self, message: &EncodedMessage) -> Result<()> { + if self + .gate() + .is_some_and(|gate| gate.blocks_outbound(&message.address)) + { + tracing::trace!(to = %message.address, "Firewall dropped outbound message"); + return Ok(()); + } + let (link, target) = self + .split(&message.address) + .with_context(|| format!("no transport for {}", message.address))?; + let mut inner = message.clone(); + inner.peer = target.clone(); + inner.address = target; + link.inner.send_encoded(&inner).await + } + + /// How the transport of `address` is connected there (`Transport::connection`). + pub(crate) fn connection(&self, address: &str) -> Option<Connection> { + let (link, target) = self.split(address)?; + link.inner.connection(&target) + } + + /// Paths' own figures, and every transport's. + pub(crate) fn stats(&self) -> TransportStats { + let mut stats = TransportStats::default(); + for link in &self.links { + stats.merge(link.inner.stats()); + } + let now = Instant::now(); + let (paths, counts) = { + let table = self.table.lock_or_recover(); + let paths: Vec<PathInfo> = table + .paths + .keys() + .map(|address| table.info(address, now)) + .collect(); + (paths, table.counts) + }; + let mut paths = self.located(paths); + paths.sort_by(|a, b| a.address.cmp(&b.address)); + stats.paths = Some(PathsStats { paths, counts }); + stats + } + + /// Sends a PING (or with `pong`, a PONG) with `nonce` to `address`, counted + /// (`traffic`). + async fn send_probe(&self, address: &str, pong: bool, nonce: u64) -> Result<()> { + let message = if pong { + ParsedMessage::Pong(nonce) + } else { + ParsedMessage::Ping(nonce) + }; + let encoded = self.encode(address, &message)?; + let sent = self.send(&encoded).await; + if let Some(traffic) = &self.traffic { + match &sent { + Ok(()) => traffic.sent(address, encoded.kind, encoded.payload_len), + Err(_) => traffic.refused(address, encoded.kind, encoded.payload_len), + } + } + sent + } + + /// Pings the paths due (`Table::due`), every `PROBE_TICK`, until dropped. Each PING + /// goes on its own, timed from when it goes: a send that takes long (a dial) neither + /// holds up the others nor counts in their round trips. + async fn probe(self: Arc<Self>) { + let mut tick = tokio::time::interval(PROBE_TICK); + tick.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay); + loop { + tick.tick().await; + let due = self + .table + .lock_or_recover() + .due(Instant::now(), random_nonce); + for (address, nonce) in due { + let paths = Arc::clone(&self); + tokio::spawn(async move { + paths + .table + .lock_or_recover() + .sending(&address, nonce, Instant::now()); + if let Err(error) = paths.send_probe(&address, false, nonce).await { + debug!(to = %address, error = %format!("{error:#}"), "Couldn't send a PING"); + paths + .table + .lock_or_recover() + .failed(&address, Instant::now()); + } + }); + } + } + } + + /// A message from `address`, handed on to `handler`: a PING answered and a PONG + /// noted first; dropped if the firewall blocks it. + async fn receive( + self: Arc<Self>, + address: String, + message: ParsedMessage, + payload_len: usize, + handler: &PathHandler, + ) -> Result<()> { + if self + .gate() + .is_some_and(|gate| gate.blocks_inbound(&address)) + { + tracing::trace!(from = %address, "Firewall dropped inbound message"); + return Ok(()); + } + match message { + ParsedMessage::Ping(nonce) => { + let paths = Arc::clone(&self); + let to = address.clone(); + tokio::spawn(async move { + if let Err(error) = paths.send_probe(&to, true, nonce).await { + debug!(to = %to, error = %format!("{error:#}"), "Couldn't answer a PING"); + } + }); + } + ParsedMessage::Pong(nonce) => { + self.table + .lock_or_recover() + .pong(&address, nonce, Instant::now()); + } + _ => {} + } + handler(address, message, payload_len).await + } + + /// Every transport's consumer, each message handed to `handler` under the sender's + /// address (PINGs and PONGs end here), and the probe task; returns when the first + /// consumer does. + pub(crate) async fn start_consumer(self: &Arc<Self>, handler: Arc<PathHandler>) -> Result<()> { + let probing = tokio::spawn(Arc::clone(self).probe()); + let mut consumers: Vec<std::pin::Pin<Box<dyn Future<Output = Result<()>> + Send>>> = + Vec::new(); + for link in &self.links { + let inner = Arc::clone(&link.inner); + let scheme = link.scheme; + let handler = Arc::clone(&handler); + let paths = Arc::clone(self); + consumers.push(Box::pin(async move { + inner + .start_consumer(Box::new(move |sender, message, payload_len| { + let paths = Arc::clone(&paths); + let handler = Arc::clone(&handler); + Box::pin(async move { + paths + .receive( + format!("{scheme}:{sender}"), + message, + payload_len, + &*handler, + ) + .await + }) + })) + .await + })); + } + let result = first(consumers) + .await + .unwrap_or_else(|| Err(anyhow!("no transports"))); + probing.abort(); + result + } + + /// When the first transport exits. + pub(crate) async fn wait_for_exit(&self) -> std::io::Result<std::process::ExitStatus> { + let waits = self + .links + .iter() + .map(|link| { + let inner = Arc::clone(&link.inner); + Box::pin(async move { inner.wait_for_exit().await }) + as std::pin::Pin<Box<dyn Future<Output = _> + Send>> + }) + .collect(); + first(waits) + .await + .unwrap_or_else(|| Err(std::io::Error::other("no transports"))) + } +} + +/// A PING's nonce: random, so a PONG can't be made up for a PING not seen. +pub(crate) fn random_nonce() -> u64 { + let mut bytes = [0u8; 8]; + if getrandom::fill(&mut bytes).is_err() { + // No randomness: a nonce still tells answers apart, if not unguessably. + return std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .map_or(0, |since| since.as_nanos() as u64); + } + u64::from_le_bytes(bytes) +} + +/// `address: error; …` for each failed send. +fn describe(failures: &[(String, anyhow::Error)]) -> String { + failures + .iter() + .map(|(address, error)| format!("{address}: {error:#}")) + .collect::<Vec<_>>() + .join("; ") +} + +/// The first of the `results` to arrive, from tasks that run until they end. +async fn first<T: Send + 'static>( + tasks: Vec<std::pin::Pin<Box<dyn Future<Output = T> + Send>>>, +) -> Option<T> { + let (sender, mut receiver) = tokio::sync::mpsc::channel(1); + for task in tasks { + let sender = sender.clone(); + tokio::spawn(async move { + let _ = sender.send(task.await).await; + }); + } + drop(sender); + receiver.recv().await +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::document::replica::Epoch; + use crate::transport::sim::SimNet; + use crate::transport::{HeadsSummary, Status}; + + /// "a" on two sim nets, as `sim` (preferred) and `other`; "b" on both too. + fn two_nets() -> (Arc<Paths>, Arc<Paths>) { + let (first, second) = (SimNet::new(1), SimNet::new(2)); + let on_both = |id: &str| { + Arc::new(Paths::new( + vec![ + ("sim", first.transport(id) as Arc<dyn Transport>), + ("other", second.transport(id) as Arc<dyn Transport>), + ], + None, + None, + )) + }; + (on_both("a"), on_both("b")) + } + + #[test] + fn addresses_by_scheme() { + let (a, _b) = two_nets(); + assert_eq!(a.local_addresses(), ["sim:a", "other:a"]); + assert_eq!(a.transports(), ["sim", "other"]); + assert_eq!(a.canonical("other:b").as_deref(), Some("other:b")); + assert_eq!(a.canonical("pigeon:b"), None); + assert_eq!(a.canonical("b"), None); // No scheme. + assert!(a.preference("sim:b") < a.preference("other:b")); + assert_eq!(a.preference("pigeon:b"), usize::MAX); + assert!(a.is_own("other:a") && !a.is_own("other:b")); + assert_eq!(a.own_address("other:b").as_deref(), Some("other:a")); + assert_eq!(a.own_address("pigeon:b"), None); + } + + #[tokio::test] + async fn seeds_resolve_to_addresses() { + let (a, _b) = two_nets(); + assert_eq!(a.resolve("other:b").await.unwrap(), "other:b"); + // Without a scheme: the first transport that resolves it. + assert_eq!(a.resolve("b").await.unwrap(), "sim:b"); + assert!(a.resolve("nobody").await.is_err()); + } + + fn paths(addresses: &[&str]) -> Vec<(String, usize)> { + addresses + .iter() + .enumerate() + .map(|(preference, address)| (address.to_string(), preference)) + .collect() + } + + /// Answers every PING due `rtt` later, as from the path it went to (`rtts`). + fn answer(table: &mut Table, now: Instant, rtts: &[(&str, u64)]) -> Instant { + let mut nonces = 0; + let due = table.due(now, || { + nonces += 1; + nonces + }); + for (address, nonce) in due { + let rtt = rtts + .iter() + .find(|(to, _)| *to == address) + .map(|(_, ms)| *ms); + if let Some(ms) = rtt { + table.pong(&address, nonce, now + Duration::from_millis(ms)); + } + } + now + PROBE_EVERY + } + + #[test] + fn the_configured_order_until_measured_then_the_faster_path() { + let mut table = Table::default(); + let mut now = Instant::now(); + let set = paths(&["libp2p:x", "mycelium:x"]); + assert_eq!(table.rank(set.clone(), now), ["libp2p:x", "mycelium:x"]); + // libp2p a bit slower: not enough to switch. + for _ in 0..8 { + now = answer(&mut table, now, &[("libp2p:x", 110), ("mycelium:x", 100)]); + } + assert_eq!(table.rank(set.clone(), now), ["libp2p:x", "mycelium:x"]); + assert_eq!(table.counts.switches, 0); + // Much slower: switched, and it stays switched while that holds. + for _ in 0..8 { + now = answer(&mut table, now, &[("libp2p:x", 300), ("mycelium:x", 100)]); + } + assert_eq!(table.rank(set.clone(), now), ["mycelium:x", "libp2p:x"]); + assert_eq!(table.rank(set.clone(), now), ["mycelium:x", "libp2p:x"]); + assert_eq!(table.counts.switches, 1); + // Small differences under SWITCH_MIN never switch. + let lan = paths(&["libp2p:y", "mycelium:y"]); + table.rank(lan.clone(), now); + for _ in 0..8 { + now = answer(&mut table, now, &[("libp2p:y", 4), ("mycelium:y", 1)]); + } + assert_eq!(table.rank(lan, now), ["libp2p:y", "mycelium:y"]); + } + + #[test] + fn unanswered_or_failing_paths_are_down() { + let mut table = Table::default(); + let mut now = Instant::now(); + let set = paths(&["libp2p:x", "mycelium:x"]); + table.rank(set.clone(), now); + // libp2p stops answering: down after UNANSWERED_DOWN PINGs, then last. + for _ in 0..=UNANSWERED_DOWN { + now = answer(&mut table, now, &[("mycelium:x", 100)]); + table.rank(set.clone(), now); + } + assert!(table.info("libp2p:x", now).down); + assert_eq!(table.rank(set.clone(), now), ["mycelium:x", "libp2p:x"]); + // It answers again: up, and measured. + now = answer(&mut table, now, &[("libp2p:x", 50), ("mycelium:x", 100)]); + assert!(!table.info("libp2p:x", now).down); + + // A failed send: down for DOWN_FOR, unless a send there works. + table.failed("mycelium:x", now); + assert!(table.info("mycelium:x", now).down); + assert!(!table.info("mycelium:x", now + DOWN_FOR).down); + table.sent("mycelium:x"); + assert!(!table.info("mycelium:x", now).down); + } + + #[test] + fn only_paths_in_use_are_pinged() { + let mut table = Table::default(); + let now = Instant::now(); + let mut nonce = 0; + let mut next = || { + nonce += 1; + nonce + }; + // One address: nothing to choose, nothing to measure. + table.rank(paths(&["libp2p:z"]), now); + assert!(table.due(now, &mut next).is_empty()); + table.rank(paths(&["libp2p:x", "mycelium:x"]), now); + assert_eq!(table.due(now, &mut next).len(), 2); + // Not again within PROBE_EVERY; a stale answer is ignored. + assert!(table.due(now + PROBE_TIMEOUT, &mut next).is_empty()); + assert_eq!(table.pong("libp2p:x", 99, now), None); + // Unused for ACTIVE_FOR: no more PINGs; for twice as long: forgotten. + assert!(table.due(now + ACTIVE_FOR, &mut next).is_empty()); + table.due(now + ACTIVE_FOR * 2, &mut next); + assert!(table.paths.is_empty() && table.chosen.is_empty()); + } + + /// A PING's round trip counts from when it goes, not from when it was found due. + #[test] + fn round_trips_count_from_the_send() { + let mut table = Table::default(); + let now = Instant::now(); + table.rank(paths(&["libp2p:x", "mycelium:x"]), now); + let due = table.due(now, || 7); + assert_eq!(due.len(), 2); + // Sent a second late (after another path's slow send). + let sent = now + Duration::from_secs(1); + table.sending("libp2p:x", 7, sent); + table.sending("libp2p:x", 8, sent + Duration::from_secs(1)); // not its PING + table.pong("libp2p:x", 7, sent + Duration::from_millis(100)); + let rtt = table.info("libp2p:x", now).rtt_ms.unwrap(); + assert!((rtt - 100.0).abs() < 1e-6, "{rtt}"); + } + + /// The node's keepalive PING on a path is its probe: paths doesn't ping it too, and + /// the PONG measures it. + #[test] + fn keepalive_pings_are_probes() { + let mut table = Table::default(); + let now = Instant::now(); + let mut nonce = 100; + let mut next = || { + nonce += 1; + nonce + }; + table.rank(paths(&["libp2p:x", "mycelium:x"]), now); + table.pinged("libp2p:x", 7, now); + let due = table.due(now, &mut next); + assert_eq!(due.len(), 1); + assert_eq!(due[0].0, "mycelium:x"); + assert!( + table + .pong("libp2p:x", 7, now + Duration::from_millis(20)) + .is_some() + ); + assert_eq!(table.counts.probes_sent, 2); + // A peer's only path is measured by its keepalives, and not pinged otherwise. + table.pinged("libp2p:z", 8, now); + assert!( + table + .pong("libp2p:z", 8, now + Duration::from_millis(30)) + .is_some() + ); + let rtt = table.info("libp2p:z", now).rtt_ms.unwrap(); + assert!((rtt - 30.0).abs() < 1e-6, "{rtt}"); + let later = now + PROBE_EVERY; + assert!( + table + .due(later, &mut next) + .iter() + .all(|(address, _)| address != "libp2p:z") + ); + // Forgotten once the keepalives stop. + table.due(now + ACTIVE_FOR * 2, &mut next); + assert!(!table.paths.contains_key("libp2p:z")); + } + + /// Blocks one address, both ways. + struct Blocking(&'static str); + + impl Gate for Blocking { + fn blocks_inbound(&self, address: &str) -> bool { + address == self.0 + } + fn blocks_outbound(&self, address: &str) -> bool { + address == self.0 + } + } + + /// The firewall drops what comes from a blocked address before anything sees it + /// (a PING isn't answered), and sends to one silently. + #[tokio::test] + async fn the_firewall_drops_both_ways() { + let net = SimNet::new(4); + let gate: Arc<dyn Gate> = Arc::new(Blocking("sim:a")); + let b = Arc::new(Paths::new( + vec![("sim", net.transport("b") as Arc<dyn Transport>)], + None, + Some(Arc::downgrade(&gate)), + )); + let a = Arc::new(Paths::new( + vec![("sim", net.transport("a") as Arc<dyn Transport>)], + None, + None, + )); + let received = |paths: &Arc<Paths>| { + let (sender, received) = tokio::sync::mpsc::unbounded_channel(); + let handler: Arc<PathHandler> = Arc::new(move |from, message, _| { + let sender = sender.clone(); + Box::pin(async move { + let _ = sender.send((from, message.kind())); + Ok(()) + }) + }); + let paths = Arc::clone(paths); + tokio::spawn(async move { paths.start_consumer(handler).await }); + received + }; + let (mut at_a, mut at_b) = (received(&a), received(&b)); + + // a's PING never reaches b, so nothing answers it. + let ping = a.encode("sim:b", &ParsedMessage::Ping(5)).unwrap(); + a.send(&ping).await.unwrap(); + // b's send to a "works", and goes nowhere. + let status = b.encode("sim:a", &some_status()).unwrap(); + b.send(&status).await.unwrap(); + tokio::time::sleep(Duration::from_millis(200)).await; + assert!(at_b.try_recv().is_err()); + assert!(at_a.try_recv().is_err()); + + // Without the gate: a PING is answered by b's paths, and goes on up there. + drop(gate); + a.send(&ping).await.unwrap(); + assert_eq!( + at_b.recv().await.unwrap(), + ("sim:a".to_owned(), crate::transport::MessageKind::Ping) + ); + assert_eq!( + at_a.recv().await.unwrap(), + ("sim:b".to_owned(), crate::transport::MessageKind::Pong) + ); + } + + /// A transport whose sends all fail. + struct Broken; + + #[async_trait::async_trait] + impl Transport for Broken { + fn local_id(&self) -> &str { + "us" + } + fn is_peer_id(&self, id: &str) -> bool { + !id.is_empty() + } + fn delivery_limit(&self) -> Duration { + Duration::from_secs(1) + } + async fn resolve_seed(&self, seed: &str) -> Result<String> { + Ok(seed.to_owned()) + } + fn encode(&self, peer: &str, message: &ParsedMessage) -> Result<EncodedMessage> { + crate::transport::mycelium::encode_message(peer, message) + } + async fn send_encoded(&self, _message: &EncodedMessage) -> Result<()> { + Err(anyhow!("broken")) + } + async fn start_consumer(&self, _handler: crate::transport::MessageHandler) -> Result<()> { + std::future::pending().await + } + async fn wait_for_exit(&self) -> std::io::Result<std::process::ExitStatus> { + std::future::pending().await + } + } + + fn some_status() -> ParsedMessage { + ParsedMessage::Status(Status { + heads: HeadsSummary::of(Vec::new()), + epoch: Epoch::GENESIS, + epoch_seq: 0, + ready: false, + syncs_with_you: false, + }) + } + + /// When the preferred transport fails, the same send goes over the next one, and + /// the failed address is ranked last for a while. + #[tokio::test] + async fn a_failed_send_goes_to_the_next_address() { + let net = SimNet::new(3); + let a = Paths::new( + vec![ + ("broken", Arc::new(Broken) as Arc<dyn Transport>), + ("sim", net.transport("a") as Arc<dyn Transport>), + ], + None, + None, + ); + let b = Arc::new(Paths::new( + vec![("sim", net.transport("b") as Arc<dyn Transport>)], + None, + None, + )); + let (sender, mut received) = tokio::sync::mpsc::unbounded_channel(); + let handler: Arc<PathHandler> = Arc::new(move |from, _, _| { + let sender = sender.clone(); + Box::pin(async move { + let _ = sender.send(from); + Ok(()) + }) + }); + tokio::spawn(async move { b.start_consumer(handler).await }); + + let addresses = vec!["sim:b".to_owned(), "broken:b".to_owned()]; + let ranked = a.rank(&addresses); + assert_eq!(ranked, ["broken:b", "sim:b"]); // The configured order. + let mut encoded = a.encode(&ranked[0], &some_status()).unwrap(); + encoded.peer = "node-b".to_owned(); + assert_eq!(a.send_first(&encoded, &ranked).await.unwrap(), "sim:b"); + assert_eq!(received.recv().await.unwrap(), "sim:a"); + assert_eq!(a.rank(&addresses), ["sim:b", "broken:b"]); + + // Without the message kept, nothing else can be tried: the transport's error. + encoded.message = None; + let error = a + .send_first(&encoded, &["broken:b".to_owned()]) + .await + .unwrap_err(); + assert_eq!(error.to_string(), "broken"); + assert!(a.send_first(&encoded, &[]).await.is_err()); + } + + /// A message sent to an address arrives under the sender's address on that + /// transport. + #[tokio::test] + async fn messages_arrive_under_the_senders_address() { + let (a, b) = two_nets(); + let (sender, mut received) = tokio::sync::mpsc::unbounded_channel(); + let handler: Arc<PathHandler> = Arc::new(move |from, message, _| { + let sender = sender.clone(); + Box::pin(async move { + let _ = sender.send((from, message.kind())); + Ok(()) + }) + }); + tokio::spawn(async move { b.start_consumer(handler).await }); + let status = some_status(); + for to in ["other:b", "sim:b"] { + let encoded = a.encode(to, &status).unwrap(); + assert_eq!(encoded.address, to); + a.send(&encoded).await.unwrap(); + } + let mut from: Vec<String> = Vec::new(); + for _ in 0..2 { + from.push(received.recv().await.unwrap().0); + } + from.sort(); + assert_eq!(from, ["other:a", "sim:a"]); + assert!(a.encode("pigeon:b", &status).is_err()); + } +} |