diff options
Diffstat (limited to 'src/swarm/links.rs')
| -rw-r--r-- | src/swarm/links.rs | 1132 |
1 files changed, 0 insertions, 1132 deletions
diff --git a/src/swarm/links.rs b/src/swarm/links.rs deleted file mode 100644 index 2234783..0000000 --- a/src/swarm/links.rs +++ /dev/null @@ -1,1132 +0,0 @@ -//! How members are linked, for a map of the swarm (the dashboard, `api::dashboard`): a -//! node asks a member with LINKS, and the member answers with a LINKS_REPLY listing its -//! neighbours and the peers it holds dormant (`own_links`), each with the transport its -//! messages go over, whether only through a relay, the round trip and whether it is -//! dormant: as the member measures them, which no other node can. -//! -//! Every node answers, whatever its build: one with a dashboard asks any member, which -//! may have none. Only members (in the member list) are answered, as the links name -//! members and where they're reached; and each at most every `ANSWER_EVERY`. Nothing -//! is written to the document: asked for while someone looks, forgotten after. -//! -//! A LINKS may subscribe (`subscribe_secs`): the member then sends a LINKS_REPLY again -//! whenever its links change (`changed`: a peer, a transport, relayed, dormant, or a -//! round trip that moved by a fifth), looked at every `WATCH_EVERY` while anyone -//! subscribes (round trips change without telling), until the subscription runs out -//! (the asker renews it with its next LINKS, so one that is gone stops it by not -//! renewing) or a LINKS with 0 ends it. -//! -//! A member the asker can't reach (on none of its transports: one on Tor alone, asked -//! from one on libp2p alone; or one that doesn't answer it, as behind NAT: as choosing -//! neighbours judges it, `topology::reaches`) is asked through a peer that can -//! (`relay_for`): a LINKS -//! naming the members (`of`) asks the receiver to ask them on the sender's behalf, and -//! to pass their links on (`relayed`). The receiver asks each itself, or through a peer -//! of its own while hops are left (`hops`), for as long as the sender asked, and holds -//! what arrives a while to send it together: the window (`batch_ms`) is the asker's for -//! the whole way (at most `MAX_BATCH`), shared out among the hops that may be left, so -//! the links reach the asker within it however long the chain (every hop holds -//! `MIN_BATCH` at least). Such a -//! subscription isn't ended by a message: it runs out, at every hop, unless the asker -//! renews it, so one that is gone (or no longer looks) stops it within a lease. -//! -//! A way of asking a member that brings none of its links (`FAILED_AFTER`: we don't -//! reach it after all, or the relay doesn't) is left out a while for that member, and -//! the next ask (the renewal) goes the next way: so the old way runs out by itself, at -//! no cost but the wait. Every hop does so for what it asks on. - -use std::collections::{HashMap, HashSet}; -use std::sync::Mutex; -use std::time::Duration; - -use tokio::time::Instant; -use tracing::debug; - -use crate::locks::MutexExt; -use crate::state::AppState; -use crate::swarm::{members, topology}; -use crate::transport::paths::{self, ClockOffset}; -use crate::transport::{ - Connection, Destination, Epoch, Link, LinksReply, LinksRequest, OtherPath, ParsedMessage, - RelayedLinks, -}; - -/// Links one reply lists at most (a member has about `--neighbours`, more if others -/// prefer it). -pub(crate) const MAX_LINKS: usize = 256; -/// Other transports in a link at most. -pub(crate) const MAX_OTHERS: usize = 8; -/// A member answers each other one at most this often. -const ANSWER_EVERY: Duration = Duration::from_secs(5); -/// Replies kept from members at most (one each), and requests out. -const MAX_KEPT: usize = 4096; -/// Subscriptions a member keeps at most, and how long one lasts at most. -const MAX_SUBSCRIBERS: usize = 64; -const MAX_SUBSCRIPTION: Duration = Duration::from_secs(300); -/// How often a member with subscribers looks whether its links changed. -const WATCH_EVERY: Duration = Duration::from_secs(2); -/// A reply to a LINKS that doesn't subscribe is taken for this long. -const REPLY_WITHIN: Duration = Duration::from_secs(30); -/// Members a relayed LINKS names at most, and that one LINKS_REPLY passes on at most. -const MAX_RELAYED: usize = 64; -/// Bytes of relayed links one LINKS_REPLY carries at most, as encoded (one member's -/// alone may be more): what a member reports is usually small (about 12 links of 150 -/// bytes), but may be up to `MAX_LINKS` links, and 64 such would make a message of -/// megabytes. -const MAX_RELAYED_BYTES: usize = 256 * 1024; -/// How long relayed links are held to be sent together, along the whole way: the -/// asker's window (at most `MAX_BATCH`), each hop holding at least `MIN_BATCH`. -pub(crate) const DEFAULT_BATCH: Duration = Duration::from_millis(250); -const MIN_BATCH: Duration = Duration::from_millis(10); -const MAX_BATCH: Duration = Duration::from_secs(5); -/// Members a peer may have us ask for it, at most: as many at once (`RELAY_BURST`), and -/// on average this many per second (`RELAY_RATE`): a dashboard asking every member -/// (`MAX_ASKED` there, 512) every 30 s needs less. The rest of a request is dropped. -const RELAY_BURST: f64 = 512.0; -const RELAY_RATE: f64 = 32.0; -/// Nodes a relayed LINKS goes through at most, the first included. -const MAX_HOPS: u8 = 3; -/// A way of asking a member (directly, or through a peer) that brought none of its -/// links for this long after we asked has failed it (`split`): the next ask goes -/// another way. More than a window (`MAX_BATCH`) along a few hops of slow transports. -const FAILED_AFTER: Duration = Duration::from_secs(20); -/// How long a way that failed a member is left out for that member. -const FAILED_KEPT: Duration = Duration::from_secs(300); - -/// The requests out and the replies in, and when each asker was last answered. -#[derive(Default)] -pub(crate) struct Links { - inner: Mutex<Inner>, - /// Someone subscribed (`answer`): `watch` looks again. - subscribed: tokio::sync::Notify, -} - -#[derive(Default)] -struct Inner { - /// Our LINKS out, by member: its nonce, and until when replies to it are taken (its - /// subscription's end, or `REPLY_WITHIN`). - asked: HashMap<String, (u64, Instant)>, - /// The latest LINKS_REPLY of each member: when, and its links. - replies: HashMap<String, (Instant, Vec<Link>)>, - /// Each member's epoch, as its latest LINKS_REPLY said (`LinksReply::epoch`). - epochs: HashMap<String, Epoch>, - /// When we last answered each member. - answered: HashMap<String, Instant>, - /// Who subscribed to our links: the nonce to answer with, and until when. - subscribers: HashMap<String, (u64, Instant)>, - /// The links subscribers were last sent (`watch`). - /// And in which epoch: a new one is news too (a dashboard shows who entered it). - published: Option<(Epoch, Vec<Link>)>, - /// Our relayed LINKS out, by the peer asked to relay: its nonce, until when replies - /// to it are taken, and the members asked for through it. - via: HashMap<String, (u64, Instant, HashSet<String>)>, - /// What we ask on others' behalf, by member asked for: who asked (`Relaying`). - relaying: HashMap<String, HashMap<String, Relaying>>, - /// Relayed links due to an asker, by asker: the members whose links changed. - due: HashMap<String, HashSet<String>>, - /// How many members each peer may still have us ask for it (`RELAY_BURST`, - /// `RELAY_RATE`), as of when. - relay_budget: HashMap<String, (f64, Instant)>, - /// How we ask each member (`split`). - routes: HashMap<String, Route>, -} - -/// How we ask a member: the way we last did, since when, whether its links came since, -/// and the ways that failed it lately (`FAILED_AFTER`), each since when. -struct Route { - /// Through this peer; None: directly. - via: Option<String>, - asked: Instant, - arrived: bool, - failed: HashMap<Option<String>, Instant>, -} - -/// One asker's relayed subscription to a member. -#[derive(Clone, Copy)] -struct Relaying { - /// The nonce to answer with. - nonce: u64, - /// Until when, if still wanted. - until: Instant, - /// How long its links are held to be sent with others. - window: Duration, - /// Asked for once (no subscription): ends with the first links passed on. - once: bool, -} - -impl Links { - /// The latest links each member replied with, and how long ago. - #[cfg_attr(not(feature = "dashboard"), allow(dead_code))] - pub(crate) fn replies(&self) -> Vec<(String, Duration, Vec<Link>)> { - let now = Instant::now(); - self.inner - .lock_or_recover() - .replies - .iter() - .map(|(member, (at, links))| { - ( - member.clone(), - now.saturating_duration_since(*at), - links.clone(), - ) - }) - .collect() - } - - /// The epoch each member was in, as its latest links said (`replies`). - #[cfg_attr(not(feature = "dashboard"), allow(dead_code))] - pub(crate) fn epochs(&self) -> HashMap<String, Epoch> { - self.inner.lock_or_recover().epochs.clone() - } - - /// Who is subscribed to our links (not run out). - #[cfg(test)] - pub(crate) fn subscribers(&self) -> Vec<String> { - let now = Instant::now(); - self.inner - .lock_or_recover() - .subscribers - .iter() - .filter(|(_, (_, until))| now < *until) - .map(|(peer, _)| peer.clone()) - .collect() - } - - /// The members we ask on others' behalf (relayed, not run out), and who asked. - #[cfg(test)] - pub(crate) fn relaying(&self) -> Vec<(String, Vec<String>)> { - let now = Instant::now(); - self.inner - .lock_or_recover() - .relaying - .iter() - .map(|(member, askers)| { - let askers: Vec<String> = askers - .iter() - .filter(|(_, relaying)| now < relaying.until) - .map(|(asker, _)| asker.clone()) - .collect(); - (member.clone(), askers) - }) - .filter(|(_, askers)| !askers.is_empty()) - .collect() - } - - /// The members our subscriptions are with, and the peers relaying for us (not run - /// out). - #[cfg_attr(not(feature = "dashboard"), allow(dead_code))] - pub(crate) fn subscribed(&self) -> Vec<String> { - let now = Instant::now(); - let inner = self.inner.lock_or_recover(); - inner - .asked - .iter() - .filter(|(_, (_, until))| now < *until) - .map(|(member, _)| member.clone()) - .chain( - inner - .via - .iter() - .filter(|(_, (_, until, _))| now < *until) - .map(|(relay, _)| relay.clone()), - ) - .collect() - } -} - -/// This node's links: its neighbours, and the peers it holds dormant, as it measures -/// them. A neighbour that goes dormant is one no more (it is unreachable now), but its -/// link stays listed, dormant, until it is heard from again: left out, the link would -/// show only as that peer last reported it, from before. -pub(crate) async fn own_links(state: &AppState) -> Vec<Link> { - let mut links = Vec::new(); - for (peer, peer_state) in state.peers_snapshot().await { - let dormant = peer_state.lock().await.liveness.is_dormant(); - if !dormant && !topology::is_neighbour(state, &peer) { - continue; - } - let destination = state.transport.destination(&peer); - let transport = destination - .as_ref() - .and_then(|destination| scheme_of(&destination.address)); - links.push(Link { - others: others(destination.as_ref(), transport.as_deref()), - transport, - relayed: destination - .as_ref() - .is_some_and(|destination| destination.connection == Some(Connection::Relayed)), - // Ranked up before down: the first is down only if all are. - down: destination - .as_ref() - .and_then(|destination| destination.paths.first()) - .is_some_and(|path| path.down), - // The path in use's (`Transport::rtt`, the first of `destination.paths`). - rtt_us: state - .transport - .rtt(&peer) - .map(|rtt| u32::try_from(rtt.as_micros()).unwrap_or(u32::MAX)), - dormant, - clock: state.transport.clock_offset(&peer), - peer, - }); - } - links.sort_by(|a, b| a.peer.cmp(&b.peer)); - links.truncate(MAX_LINKS); - links -} - -/// The scheme of `address` (`mycelium:…`): its transport (`paths::scheme_of`), if it -/// has one. -fn scheme_of(address: &str) -> Option<String> { - Some(paths::scheme_of(address)) - .filter(|scheme| !scheme.is_empty()) - .map(str::to_owned) -} - -/// The transports besides `used` the peer proved itself at, with a path up: each with -/// the round trip of its fastest path, if any is measured, and whether all of them go -/// through a relay. -fn others(destination: Option<&Destination>, used: Option<&str>) -> Vec<OtherPath> { - let mut others: Vec<OtherPath> = Vec::new(); - for path in destination.map_or(&[][..], |destination| &destination.paths[..]) { - let Some(transport) = scheme_of(&path.address) else { - continue; - }; - if path.down || used == Some(transport.as_str()) { - continue; - } - let rtt_us = path - .rtt_ms - .filter(|ms| ms.is_finite() && *ms >= 0.0) - .map(|ms| (ms * 1000.0).min(f64::from(u32::MAX)) as u32); - let relayed = path.connection == Some(Connection::Relayed); - match others.iter_mut().find(|other| other.transport == transport) { - Some(other) => { - other.rtt_us = match (other.rtt_us, rtt_us) { - (Some(a), Some(b)) => Some(a.min(b)), - (a, b) => a.or(b), - }; - other.relayed &= relayed; - } - None => others.push(OtherPath { - transport, - rtt_us, - relayed, - }), - } - } - others.sort_by(|a, b| a.transport.cmp(&b.transport)); - others.truncate(MAX_OTHERS); - others -} - -/// Asks `member` for its links (a LINKS), and with `subscribe` for its changes for -/// that long; its replies land in `AppState::links`. -#[cfg_attr(not(feature = "dashboard"), allow(dead_code))] -pub(crate) async fn ask(state: &AppState, member: &str, subscribe: Duration) { - let nonce = { - let now = Instant::now(); - let mut inner = state.links.inner.lock_or_recover(); - if inner.asked.len() >= MAX_KEPT && !inner.asked.contains_key(member) { - inner.asked.clear(); - } - let until = now + subscribe.max(REPLY_WITHIN); - // The same nonce while its replies are taken: a renewal doesn't drop a reply on - // its way under the last one (in a chain of relays, every hop renews at once, - // `relay`), as `ask_via` keeps one per relay. - match inner.asked.get_mut(member) { - Some((nonce, taken)) if now < *taken => { - *taken = (*taken).max(until); - *nonce - } - _ => { - let nonce = crate::transport::paths::random_nonce(); - inner.asked.insert(member.to_owned(), (nonce, until)); - nonce - } - } - }; - let (epoch, epoch_seq) = state.epoch_stamp(); - let request = ParsedMessage::Links(LinksRequest { - epoch, - epoch_seq, - nonce, - subscribe_secs: u32::try_from(subscribe.as_secs()).unwrap_or(u32::MAX), - of: Vec::new(), - batch_ms: 0, - hops: 0, - }); - if let Err(error) = state.transport.send_message(member, &request).await { - debug!(member = %member, error = %format!("{error:#}"), "LINKS failed"); - } -} - -/// Asks `members` for their links (`ask`): those we can reach ourselves, the others -/// through peers that can (`relay_for`, `ask_via`), with our window for what they pass -/// on (`DEFAULT_BATCH`). -#[cfg_attr(not(feature = "dashboard"), allow(dead_code))] -pub(crate) async fn ask_all(state: &AppState, members: Vec<String>, subscribe: Duration) { - let (direct, through) = split(state, members, None, true).await; - for member in direct { - let state = state.clone(); - tokio::spawn(async move { ask(&state, &member, subscribe).await }); - } - for (relay, members) in through { - ask_via( - state, - &relay, - members, - subscribe, - DEFAULT_BATCH, - MAX_HOPS - 1, - ) - .await; - } -} - -/// `members` we can reach ourselves, and the others by the peer to ask them through -/// (none for those no peer leads to, nor without `through_peers`); not `exclude` (who -/// asked us). A way that brought none of a member's links since we last asked it that -/// way (`FAILED_AFTER`) is left out for a while (`FAILED_KEPT`): directly, the next -/// relay, and so on; once every way failed, they are all tried again. -async fn split( - state: &AppState, - members: Vec<String>, - exclude: Option<&str>, - through_peers: bool, -) -> (Vec<String>, HashMap<String, Vec<String>>) { - let mut direct = Vec::new(); - let mut through: HashMap<String, Vec<String>> = HashMap::new(); - for member in members { - if member == state.local_id() || Some(member.as_str()) == exclude { - continue; - } - let failed = failed_ways(state, &member); - let mut via = way(state, &member, exclude, through_peers, &failed).await; - if via.is_none() && !failed.is_empty() { - forget_failed(state, &member); - via = way(state, &member, exclude, through_peers, &HashMap::new()).await; - } - let Some(via) = via else { - continue; - }; - asked_by(state, &member, via.clone()); - match via { - None => direct.push(member), - Some(relay) => through.entry(relay).or_default().push(member), - } - } - (direct, through) -} - -/// How to ask `member`, but the `failed` ways: Some(None) directly, Some(Some(peer)) -/// through `peer` (if `through_peers`); None if no way is left. -async fn way( - state: &AppState, - member: &str, - exclude: Option<&str>, - through_peers: bool, - failed: &HashMap<Option<String>, Instant>, -) -> Option<Option<String>> { - if !failed.contains_key(&None) && topology::reaches(state, member).await { - return Some(None); - } - if !through_peers { - return None; - } - relay_for(state, member, exclude, failed).await.map(Some) -} - -/// The ways that failed `member` lately; the one we last asked it by among them if it -/// brought nothing within `FAILED_AFTER`. -fn failed_ways(state: &AppState, member: &str) -> HashMap<Option<String>, Instant> { - let now = Instant::now(); - let mut inner = state.links.inner.lock_or_recover(); - let Some(route) = inner.routes.get_mut(member) else { - return HashMap::new(); - }; - route - .failed - .retain(|_, since| now.saturating_duration_since(*since) < FAILED_KEPT); - if !route.arrived && now.saturating_duration_since(route.asked) >= FAILED_AFTER { - debug!(member = %member, via = ?route.via, "LINKS brought nothing that way; asking another"); - route.failed.insert(route.via.clone(), now); - } - route.failed.clone() -} - -/// Every way may be tried again for `member`. -fn forget_failed(state: &AppState, member: &str) { - if let Some(route) = state.links.inner.lock_or_recover().routes.get_mut(member) { - route.failed.clear(); - } -} - -/// We ask `member` `via` now: the wait for its links starts over, unless we asked it so -/// already and are still waiting. -fn asked_by(state: &AppState, member: &str, via: Option<String>) { - let now = Instant::now(); - let mut inner = state.links.inner.lock_or_recover(); - if inner.routes.len() >= MAX_KEPT && !inner.routes.contains_key(member) { - inner.routes.clear(); - } - let route = inner - .routes - .entry(member.to_owned()) - .or_insert_with(|| Route { - via: via.clone(), - asked: now, - arrived: false, - failed: HashMap::new(), - }); - if route.via != via || route.arrived { - route.via = via; - route.asked = now; - route.arrived = false; - } -} - -/// The peer to ask `member` through, which we can't reach: an active member we reach -/// (that answers us) that, by the member list, is on a transport `member` is on; else one on a transport -/// we aren't on (it may know one that is). Of several, the one with the lowest pair score -/// with `member` (`topology::pair_score`), so askers pick alike. Not `exclude`, nor one -/// that failed `member` (`failed`). -async fn relay_for( - state: &AppState, - member: &str, - exclude: Option<&str>, - failed: &HashMap<Option<String>, Instant>, -) -> Option<String> { - let theirs = topology::transports_of(state, member); - let ours = topology::transports_of(state, state.local_id()); - let mut best: Option<(bool, u64, String)> = None; - for (peer, peer_state) in state.peers_snapshot().await { - if peer == member - || Some(peer.as_str()) == exclude - || failed.contains_key(&Some(peer.clone())) - || !state.transport.can_reach(&peer) - { - continue; - } - { - let peer_state = peer_state.lock().await; - if peer_state.is_unreachable() - || peer_state.liveness.is_dormant() - || !matches!(peer_state.standing, crate::state::Standing::Member { .. }) - { - continue; - } - } - let on = topology::transports_of(state, &peer); - let shares = !on.is_disjoint(&theirs); - if !shares && on.is_subset(&ours) { - continue; - } - let score = topology::pair_score(&peer, member); - // Sharing a transport with the member first, then the lowest score. - let better = best.as_ref().is_none_or(|(best_shares, best_score, _)| { - (shares, std::cmp::Reverse(score)) > (*best_shares, std::cmp::Reverse(*best_score)) - }); - if better { - best = Some((shares, score, peer)); - } - } - best.map(|(_, _, peer)| peer) -} - -/// Asks `relay` to ask `members` on our behalf (a LINKS naming them), with `subscribe`, -/// passing their links on within `window`, through `hops` more nodes at most: in LINKS -/// of `MAX_RELAYED` members at most. One nonce per relay while its replies are taken, so -/// its replies for every member we asked it for are taken. -async fn ask_via( - state: &AppState, - relay: &str, - members: Vec<String>, - subscribe: Duration, - window: Duration, - hops: u8, -) { - let nonce = { - let now = Instant::now(); - let mut inner = state.links.inner.lock_or_recover(); - if inner.via.len() >= MAX_KEPT && !inner.via.contains_key(relay) { - inner.via.retain(|_, (_, until, _)| now < *until); - } - let until = now + subscribe.max(REPLY_WITHIN); - let entry = inner.via.entry(relay.to_owned()).or_insert_with(|| { - ( - crate::transport::paths::random_nonce(), - until, - HashSet::new(), - ) - }); - if now >= entry.1 { - entry.0 = crate::transport::paths::random_nonce(); - entry.2.clear(); - } - entry.1 = entry.1.max(until); - entry.2.extend(members.iter().cloned()); - entry.0 - }; - let (epoch, epoch_seq) = state.epoch_stamp(); - for members in members.chunks(MAX_RELAYED) { - let request = ParsedMessage::Links(LinksRequest { - epoch, - epoch_seq, - nonce, - subscribe_secs: u32::try_from(subscribe.as_secs()).unwrap_or(u32::MAX), - of: members.to_vec(), - batch_ms: u32::try_from(window.as_millis()).unwrap_or(u32::MAX), - hops, - }); - if let Err(error) = state.transport.send_message(relay, &request).await { - debug!(relay = %relay, error = %format!("{error:#}"), "Relayed LINKS failed"); - } - } -} - -/// Ends our subscription with `member` (a LINKS with 0); its replies aren't taken any -/// more. A peer relaying for us isn't told: what it asks for us runs out by itself. -#[cfg_attr(not(feature = "dashboard"), allow(dead_code))] -pub(crate) async fn unsubscribe(state: &AppState, member: &str) { - let nonce = { - let mut inner = state.links.inner.lock_or_recover(); - if inner.via.remove(member).is_some() && !inner.asked.contains_key(member) { - return; - } - match inner.asked.remove(member) { - Some((nonce, _)) => nonce, - None => return, - } - }; - let (epoch, epoch_seq) = state.epoch_stamp(); - let request = ParsedMessage::Links(LinksRequest { - epoch, - epoch_seq, - nonce, - subscribe_secs: 0, - of: Vec::new(), - batch_ms: 0, - hops: 0, - }); - if let Err(error) = state.transport.send_message(member, &request).await { - debug!(member = %member, error = %format!("{error:#}"), "LINKS failed"); - } -} - -/// A LINKS from `peer`: answered if it's a member, and not answered lately. -pub(crate) async fn answer(state: &AppState, peer: &str, request: LinksRequest) { - // From the neighbours' view, kept as the member list changes: no document read per - // request. - let member = match crate::swarm::topology::listed(state, peer) { - Some(listed) => listed, - None => { - let replica = state.doc.read().await; - members::members(&replica.doc) - .iter() - .any(|member| member == peer) - } - }; - if !member { - debug!(peer = %peer, "Not answering LINKS from a peer that isn't a member"); - return; - } - if !request.of.is_empty() { - relay(state, peer, request).await; - return; - } - { - let now = Instant::now(); - let mut inner = state.links.inner.lock_or_recover(); - // A subscription is taken, renewed or ended even when the reply is due later. - if request.subscribe_secs == 0 { - inner.subscribers.remove(peer); - } else if inner.subscribers.len() < MAX_SUBSCRIBERS || inner.subscribers.contains_key(peer) - { - let lasts = - Duration::from_secs(u64::from(request.subscribe_secs)).min(MAX_SUBSCRIPTION); - inner - .subscribers - .insert(peer.to_owned(), (request.nonce, now + lasts)); - state.links.subscribed.notify_one(); - } - if inner - .answered - .get(peer) - .is_some_and(|at| now.saturating_duration_since(*at) < ANSWER_EVERY) - { - return; - } - if inner.answered.len() >= MAX_KEPT { - inner - .answered - .retain(|_, at| now.saturating_duration_since(*at) < ANSWER_EVERY); - } - inner.answered.insert(peer.to_owned(), now); - } - let (epoch, epoch_seq) = state.epoch_stamp(); - let reply = ParsedMessage::LinksReply(LinksReply { - epoch, - epoch_seq, - nonce: request.nonce, - links: own_links(state).await, - relayed: Vec::new(), - }); - if let Err(error) = state.transport.send_message(peer, &reply).await { - debug!(peer = %peer, error = %format!("{error:#}"), "LINKS_REPLY failed"); - } -} - -/// A relayed LINKS from `asker` (`answer`): the members it names are asked on its -/// behalf, ourselves or through a peer of ours while hops are left, and what we know of -/// them already is passed on at once (within its window); the rest as it arrives. -async fn relay(state: &AppState, asker: &str, request: LinksRequest) { - let now = Instant::now(); - let lease = Duration::from_secs(u64::from(request.subscribe_secs)).min(MAX_SUBSCRIPTION); - let once = lease.is_zero(); - // What is left of the asker's window, shared out among this hop and those that may - // follow: ours is a share, the rest goes on. - let left = Duration::from_millis(u64::from(request.batch_ms)).min(MAX_BATCH); - let window = (left / (u32::from(request.hops) + 1)).max(MIN_BATCH); - let rest = left.saturating_sub(window); - let mut members: Vec<String> = request - .of - .into_iter() - .filter(|member| member != asker && member != state.local_id()) - .take(MAX_RELAYED) - .collect(); - let known: Vec<String> = { - let mut inner = state.links.inner.lock_or_recover(); - // Within the asker's budget: each member costs one. - if inner.relay_budget.len() >= MAX_KEPT && !inner.relay_budget.contains_key(asker) { - inner.relay_budget.retain(|_, (budget, at)| { - *budget + now.saturating_duration_since(*at).as_secs_f64() * RELAY_RATE - < RELAY_BURST - }); - } - let (budget, at) = inner - .relay_budget - .entry(asker.to_owned()) - .or_insert((RELAY_BURST, now)); - *budget = (*budget + now.saturating_duration_since(*at).as_secs_f64() * RELAY_RATE) - .min(RELAY_BURST); - *at = now; - let allowed = (*budget as usize).min(members.len()); - *budget -= allowed as f64; - if allowed < members.len() { - debug!(asker = %asker, asked = members.len(), allowed, "Relayed LINKS over its budget; asking for part of it"); - members.truncate(allowed); - } - for member in &members { - if inner.relaying.len() >= MAX_KEPT && !inner.relaying.contains_key(member) { - inner - .relaying - .retain(|_, askers| askers.values().any(|relaying| now < relaying.until)); - } - let askers = inner.relaying.entry(member.clone()).or_default(); - askers.retain(|_, relaying| now < relaying.until); - if askers.len() >= MAX_SUBSCRIBERS && !askers.contains_key(asker) { - continue; - } - askers.insert( - asker.to_owned(), - Relaying { - nonce: request.nonce, - // Asked once: for as long as its reply is awaited. - until: now + if once { REPLY_WITHIN } else { lease }, - window, - once, - }, - ); - } - members - .iter() - .filter(|member| { - inner - .replies - .get(*member) - .is_some_and(|(at, _)| now.saturating_duration_since(*at) < REPLY_WITHIN) - }) - .cloned() - .collect() - }; - for member in known { - forward(state, &member); - } - let (direct, through) = split(state, members, Some(asker), request.hops > 0).await; - for member in direct { - // Asked once while we subscribe anyway (our own dashboard, or another asker's): - // what comes is passed on, and a LINKS with 0 would end that subscription. - let subscribed = once - && state - .links - .inner - .lock_or_recover() - .asked - .get(&member) - .is_some_and(|(_, until)| now + REPLY_WITHIN < *until); - if !subscribed { - ask(state, &member, lease).await; - } - } - for (next, members) in through { - ask_via(state, &next, members, lease, rest, request.hops - 1).await; - } -} - -/// `member`'s links changed (or arrived): due to those who asked us for them -/// (`relaying`), each sent with what else is due to it at the end of its window. -fn forward(state: &AppState, member: &str) { - let now = Instant::now(); - let mut starting = Vec::new(); - { - let mut inner = state.links.inner.lock_or_recover(); - let askers: Vec<(String, Duration)> = match inner.relaying.get_mut(member) { - Some(askers) => { - askers.retain(|_, relaying| now < relaying.until); - askers - .iter() - .map(|(asker, relaying)| (asker.clone(), relaying.window)) - .collect() - } - None => return, - }; - for (asker, window) in askers { - let due = inner.due.entry(asker.clone()).or_default(); - if due.is_empty() { - starting.push((asker, window)); - } - due.insert(member.to_owned()); - } - } - for (asker, window) in starting { - let state = state.clone(); - tokio::spawn(async move { - tokio::time::sleep(window).await; - pass_on(&state, &asker).await; - }); - } -} - -/// What is due to `asker`: the links of the members it asked for, as we have them, in -/// as few LINKS_REPLYs as `MAX_RELAYED` allows (one per nonce it asked with). -async fn pass_on(state: &AppState, asker: &str) { - let now = Instant::now(); - let mut by_nonce: HashMap<u64, Vec<RelayedLinks>> = HashMap::new(); - { - let mut inner = state.links.inner.lock_or_recover(); - let due = inner.due.remove(asker).unwrap_or_default(); - for member in due { - let Some(relaying) = inner - .relaying - .get(&member) - .and_then(|askers| askers.get(asker)) - .copied() - .filter(|relaying| now < relaying.until) - else { - continue; - }; - let Some((at, links)) = inner.replies.get(&member) else { - continue; - }; - let Some(epoch) = inner.epochs.get(&member).copied() else { - continue; - }; - let age_ms = - u32::try_from(now.saturating_duration_since(*at).as_millis()).unwrap_or(u32::MAX); - by_nonce - .entry(relaying.nonce) - .or_default() - .push(RelayedLinks { - member: member.clone(), - age_ms, - epoch, - links: links.clone(), - }); - if relaying.once - && let Some(askers) = inner.relaying.get_mut(&member) - { - askers.remove(asker); - } - } - } - let (epoch, epoch_seq) = state.epoch_stamp(); - for (nonce, relayed) in by_nonce { - for part in by_size(relayed) { - let reply = ParsedMessage::LinksReply(LinksReply { - epoch, - epoch_seq, - nonce, - links: Vec::new(), - relayed: part, - }); - if let Err(error) = state.transport.send_message(asker, &reply).await { - debug!(asker = %asker, error = %format!("{error:#}"), "Relayed LINKS_REPLY failed"); - } - } - } -} - -/// `relayed` in parts for a LINKS_REPLY each: `MAX_RELAYED` members' links at most, and -/// `MAX_RELAYED_BYTES` as encoded (but for one member's alone). -fn by_size(relayed: Vec<RelayedLinks>) -> Vec<Vec<RelayedLinks>> { - let mut parts: Vec<Vec<RelayedLinks>> = Vec::new(); - let mut bytes = 0; - for links in relayed { - let size = rmp_serde::encode::to_vec(&links).map_or(0, |encoded| encoded.len()); - let full = parts.last().is_none_or(|part| { - part.len() >= MAX_RELAYED || (!part.is_empty() && bytes + size > MAX_RELAYED_BYTES) - }); - if full { - parts.push(Vec::new()); - bytes = 0; - } - bytes += size; - parts.last_mut().expect("just made").push(links); - } - parts -} - -/// A LINKS_REPLY from `peer`: its own links kept if it answers our latest LINKS to it, -/// and those it relays if it answers our latest relayed one, while their replies are -/// taken (a subscription's go on); then passed on to whoever asked us for them. -pub(crate) fn received(state: &AppState, peer: &str, reply: LinksReply) { - let now = Instant::now(); - let mut arrived = Vec::new(); - { - let mut inner = state.links.inner.lock_or_recover(); - let own = inner - .asked - .get(peer) - .is_some_and(|(nonce, until)| *nonce == reply.nonce && now < *until); - let via: Option<HashSet<String>> = inner - .via - .get(peer) - .filter(|(nonce, until, _)| *nonce == reply.nonce && now < *until) - .map(|(_, _, members)| members.clone()); - let mut keep = |member: String, age: Duration, epoch: Epoch, mut links: Vec<Link>| { - links.truncate(MAX_LINKS); - for link in &mut links { - link.others.truncate(MAX_OTHERS); - } - if inner.replies.len() >= MAX_KEPT && !inner.replies.contains_key(&member) { - inner.replies.clear(); - inner.epochs.clear(); - } - let at = now.checked_sub(age).unwrap_or(now); - // Arrived, if older news: the way works. - if let Some(route) = inner.routes.get_mut(&member) { - route.arrived = true; - } - // Not older news over newer. - let known = inner.replies.get(&member); - if known.is_some_and(|(known, _)| *known > at) { - return; - } - // The same links as we have, lately: no news, so not passed on. A copy that - // went round between relays (each asking the other for the member) comes - // back a little newer each time, as its age counts from when each sent it: - // passed on, it would go round for good. - let news = known.is_none_or(|(known_at, known_links)| { - *known_links != links || now.saturating_duration_since(*known_at) >= REPLY_WITHIN - }) || inner.epochs.get(&member) != Some(&epoch); - inner.replies.insert(member.clone(), (at, links)); - inner.epochs.insert(member.clone(), epoch); - if news { - arrived.push(member); - } - }; - if own { - keep(peer.to_owned(), Duration::ZERO, reply.epoch, reply.links); - } - if let Some(members) = via { - for relayed in reply.relayed.into_iter().take(MAX_RELAYED) { - if members.contains(&relayed.member) { - keep( - relayed.member, - Duration::from_millis(u64::from(relayed.age_ms)), - relayed.epoch, - relayed.links, - ); - } - } - } - } - for member in arrived { - forward(state, &member); - } -} - -/// For subscribers (`answer`): every `WATCH_EVERY`, while there are any, this node's -/// links; sent to each of them when they changed since they were last sent. Without -/// any, it waits for one. Runs for the node's life. -pub(crate) async fn watch(state: AppState) { - let mut every = tokio::time::interval(WATCH_EVERY); - every.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay); - loop { - every.tick().await; - let subscribers: Vec<(String, u64)> = { - let now = Instant::now(); - let mut inner = state.links.inner.lock_or_recover(); - inner.subscribers.retain(|_, (_, until)| now < *until); - if inner.subscribers.is_empty() { - inner.published = None; - } - inner - .subscribers - .iter() - .map(|(peer, (nonce, _))| (peer.clone(), *nonce)) - .collect() - }; - if subscribers.is_empty() { - // A new subscriber was just answered (`answer`, or within `ANSWER_EVERY`): - // the next look is due an interval later. - state.links.subscribed.notified().await; - every.reset(); - continue; - } - let links = own_links(&state).await; - let (epoch, epoch_seq) = state.epoch_stamp(); - { - let mut inner = state.links.inner.lock_or_recover(); - if inner - .published - .as_ref() - .is_some_and(|(published_in, published)| { - *published_in == epoch && !changed(published, &links) - }) - { - continue; - } - inner.published = Some((epoch, links.clone())); - } - for (peer, nonce) in subscribers { - let reply = ParsedMessage::LinksReply(LinksReply { - epoch, - epoch_seq, - nonce, - links: links.clone(), - relayed: Vec::new(), - }); - if let Err(error) = state.transport.send_message(&peer, &reply).await { - debug!(peer = %peer, error = %format!("{error:#}"), "LINKS_REPLY failed"); - } - } - } -} - -/// Whether `new` says something `old` didn't: another peer, transport, relay or -/// dormancy, a round trip measured or moved by more than a fifth (and 1 ms), or a clock -/// measured or moved beyond both bounds. Round trips and clock samples wobble; small -/// moves aren't news. -pub(crate) fn changed(old: &[Link], new: &[Link]) -> bool { - let clock_moved = |a: Option<ClockOffset>, b: Option<ClockOffset>| match (a, b) { - (Some(a), Some(b)) => { - a.offset_ms.abs_diff(b.offset_ms) > a.within_ms.saturating_add(b.within_ms) - } - (a, b) => a.is_some() != b.is_some(), - }; - let moved = |a: Option<u32>, b: Option<u32>| match (a, b) { - (Some(a), Some(b)) => { - let by = a.abs_diff(b); - by > 1000 && by > a.max(b) / 5 - } - (a, b) => a.is_some() != b.is_some(), - }; - old.len() != new.len() - || old.iter().zip(new).any(|(a, b)| { - a.peer != b.peer - || a.transport != b.transport - || a.relayed != b.relayed - || a.down != b.down - || a.dormant != b.dormant - || moved(a.rtt_us, b.rtt_us) - || clock_moved(a.clock, b.clock) - || a.others.len() != b.others.len() - || a.others.iter().zip(&b.others).any(|(x, y)| { - x.transport != y.transport - || x.relayed != y.relayed - || moved(x.rtt_us, y.rtt_us) - }) - }) -} - -#[cfg(test)] -mod tests { - use super::*; - - #[test] - fn relayed_links_are_sent_in_parts_by_count_and_size() { - let relayed = |links: usize| RelayedLinks { - member: "m".into(), - age_ms: 0, - epoch: Epoch { - generation: 1, - id: [0; 32], - }, - links: (0..links) - .map(|i| link(&format!("peer-{i:0>70}"), Some(1500))) - .collect(), - }; - // Small: by count. - let parts = by_size((0..130).map(|_| relayed(2)).collect()); - assert_eq!(parts.iter().map(Vec::len).collect::<Vec<_>>(), [64, 64, 2]); - // Large: by size, each part within the bytes but for one alone. - let big = rmp_serde::encode::to_vec(&relayed(MAX_LINKS)) - .unwrap() - .len(); - assert!(20 * big > MAX_RELAYED_BYTES, "{big}"); - let parts = by_size((0..20).map(|_| relayed(MAX_LINKS)).collect()); - assert_eq!(parts.iter().map(Vec::len).sum::<usize>(), 20); - for part in &parts { - let bytes: usize = part - .iter() - .map(|links| rmp_serde::encode::to_vec(links).unwrap().len()) - .sum(); - assert!(part.len() == 1 || bytes <= MAX_RELAYED_BYTES, "{bytes}"); - } - assert!(parts.len() > 1); - } - - fn link(peer: &str, rtt_us: Option<u32>) -> Link { - Link { - peer: peer.into(), - transport: Some("libp2p".into()), - relayed: false, - down: false, - rtt_us, - dormant: false, - others: Vec::new(), - clock: None, - } - } - - #[test] - fn only_news_is_a_change() { - let old = [link("a", Some(10_000))]; - assert!(!changed(&old, &[link("a", Some(11_000))])); - assert!(changed(&old, &[link("a", Some(13_000))])); - assert!(changed(&old, &[link("a", None)])); - assert!(changed(&old, &[link("b", Some(10_000))])); - assert!(changed(&old, &[])); - let mut dormant = link("a", Some(10_000)); - dormant.dormant = true; - assert!(changed(&old, &[dormant])); - let mut down = link("a", Some(10_000)); - down.down = true; - assert!(changed(&old, &[down])); - // Small round trips: a millisecond's wobble isn't news. - assert!(!changed(&[link("a", Some(400))], &[link("a", Some(1_300))])); - // A clock: news when measured, and when it moved beyond both bounds. - let clocked = |offset_ms| Link { - clock: Some(ClockOffset { - offset_ms, - within_ms: 50, - }), - ..link("a", Some(10_000)) - }; - assert!(changed(&old, &[clocked(0)])); - assert!(!changed(&[clocked(0)], &[clocked(80)])); - assert!(changed(&[clocked(0)], &[clocked(5_000)])); - } -} |