diff options
Diffstat (limited to 'src/swarm')
| -rw-r--r-- | src/swarm/links.rs | 585 | ||||
| -rw-r--r-- | src/swarm/members.rs | 65 | ||||
| -rw-r--r-- | src/swarm/peer/worker.rs | 10 | ||||
| -rw-r--r-- | src/swarm/relay.rs | 56 | ||||
| -rw-r--r-- | src/swarm/topology/mod.rs | 59 |
5 files changed, 713 insertions, 62 deletions
diff --git a/src/swarm/links.rs b/src/swarm/links.rs index b84f57a..7f8f0b0 100644 --- a/src/swarm/links.rs +++ b/src/swarm/links.rs @@ -15,8 +15,20 @@ //! 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) 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. -use std::collections::HashMap; +use std::collections::{HashMap, HashSet}; use std::sync::Mutex; use std::time::Duration; @@ -28,7 +40,7 @@ use crate::state::AppState; use crate::swarm::{members, topology}; use crate::transport::paths::{self, ClockOffset}; use crate::transport::{ - Connection, Destination, Link, LinksReply, LinksRequest, OtherPath, ParsedMessage, + Connection, Destination, Link, LinksReply, LinksRequest, OtherPath, ParsedMessage, RelayedLinks, }; /// Links one reply lists at most (a member has about `--neighbours`, more if others @@ -47,6 +59,25 @@ const MAX_SUBSCRIPTION: Duration = Duration::from_secs(300); 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; /// The requests out and the replies in, and when each asker was last answered. #[derive(Default)] @@ -69,6 +100,29 @@ struct Inner { subscribers: HashMap<String, (u64, Instant)>, /// The links subscribers were last sent (`watch`). published: Option<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)>, +} + +/// 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 { @@ -90,16 +144,57 @@ impl Links { .collect() } - /// The members our subscriptions are with (not run out). - #[cfg_attr(not(feature = "dashboard"), allow(dead_code))] - pub(crate) fn subscribed(&self) -> Vec<String> { + /// 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() } } @@ -194,33 +289,185 @@ fn others(destination: Option<&Destination>, used: Option<&str>) -> Vec<OtherPat /// 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 = crate::transport::paths::random_nonce(); - { + 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 = Instant::now() + subscribe.max(REPLY_WITHIN); - inner.asked.insert(member.to_owned(), (nonce, until)); - } + 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).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); not `exclude` (who asked us). +async fn split( + state: &AppState, + members: Vec<String>, + exclude: Option<&str>, +) -> (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; + } + if state.transport.can_reach(&member) { + direct.push(member); + } else if let Some(relay) = relay_for(state, &member, exclude).await { + through.entry(relay).or_default().push(member); + } + } + (direct, through) +} + +/// The peer to ask `member` through, which we can't reach: an active member we reach +/// 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`. +async fn relay_for(state: &AppState, member: &str, exclude: Option<&str>) -> 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 || !state.transport.can_reach(&peer) { + continue; + } + { + let peer_state = peer_state.lock().await; + if 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. +/// 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, @@ -232,6 +479,9 @@ pub(crate) async fn unsubscribe(state: &AppState, member: &str) { 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"); @@ -250,6 +500,10 @@ pub(crate) async fn answer(state: &AppState, peer: &str, request: LinksRequest) 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(); @@ -285,33 +539,276 @@ pub(crate) async fn answer(state: &AppState, peer: &str, request: LinksRequest) 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 LINKS_REPLY from `peer`: kept if it answers our latest LINKS to it, while its -/// replies are taken (a subscription's go on). -pub(crate) fn received(state: &AppState, peer: &str, mut reply: LinksReply) { - let mut inner = state.links.inner.lock_or_recover(); - let taken = inner - .asked - .get(peer) - .is_some_and(|(nonce, until)| *nonce == reply.nonce && Instant::now() < *until); - if !taken { - return; +/// 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); } - reply.links.truncate(MAX_LINKS); - for link in &mut reply.links { - link.others.truncate(MAX_OTHERS); + let (direct, through) = split(state, members, Some(asker)).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; + } } - if inner.replies.len() >= MAX_KEPT && !inner.replies.contains_key(peer) { - inner.replies.clear(); + if request.hops > 0 { + 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 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, + 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, 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(); + } + let at = now.checked_sub(age).unwrap_or(now); + // Not older news over newer. + if inner + .replies + .get(&member) + .is_some_and(|(known, _)| *known > at) + { + return; + } + inner.replies.insert(member.clone(), (at, links)); + arrived.push(member); + }; + if own { + keep(peer.to_owned(), Duration::ZERO, 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.links, + ); + } + } + } + } + for member in arrived { + forward(state, &member); } - inner - .replies - .insert(peer.to_owned(), (Instant::now(), reply.links)); } /// For subscribers (`answer`): every `WATCH_EVERY`, while there are any, this node's @@ -361,6 +858,7 @@ pub(crate) async fn watch(state: AppState) { 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"); @@ -409,6 +907,35 @@ fn changed(old: &[Link], new: &[Link]) -> bool { 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, + 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(), diff --git a/src/swarm/members.rs b/src/swarm/members.rs index 05c6501..a534c04 100644 --- a/src/swarm/members.rs +++ b/src/swarm/members.rs @@ -10,7 +10,8 @@ //! //! Besides its tier, an entry holds the alias a node gives itself (`--alias`), a name //! for people reading /peers and metrics (not unique and never an identity: the public -//! key stays that), where it can be reached (`addresses`, as its transport gives them: +//! key stays that), where people reach its dashboard (`--dashboard-url`, `dashboard`), +//! where it can be reached (`addresses`, as its transport gives them: //! hints for the others' transports), and an `alive` stamp the node renews now and //! then: the members renewed within `--electorate-window` are the electorate of a //! remake. @@ -114,6 +115,34 @@ fn set_alias(doc: &mut impl Transactable, pk: &str, alias: Option<&str>) -> Resu } } +fn dashboard_path(pk: &str) -> Vec<String> { + let mut path = member_path(pk); + path.push("dashboard".to_owned()); + path +} + +/// Where people reach `pk`'s dashboard, if it says. +pub(crate) fn dashboard(doc: &Document, pk: &str) -> Option<String> { + match store::read_path(doc, &dashboard_path(pk)) { + Ok(Some(JsonValue::String(url))) => Some(url), + _ => None, + } +} + +fn set_dashboard( + doc: &mut impl Transactable, + pk: &str, + url: Option<&str>, +) -> Result<(), StoreError> { + match url { + Some(url) => store::write_path(doc, &dashboard_path(pk), &json!(url)), + None => match store::delete_path(doc, &dashboard_path(pk), false) { + Err(StoreError::NotFound) => Ok(()), + result => result, + }, + } +} + fn addresses_path(pk: &str) -> Vec<String> { let mut path = member_path(pk); path.push("addresses".to_owned()); @@ -260,8 +289,9 @@ const RENEWAL_MOST_OFTEN: Duration = Duration::from_secs(3600); pub(crate) const MIN_ELECTORATE_WINDOW: Duration = Duration::from_secs(7 * RENEWAL_MOST_OFTEN.as_secs()); -/// Registers this node if it is ready and its entry is missing, keeps its alias as -/// configured (set, changed or removed after a restart with another `--alias`) and its +/// Registers this node if it is ready and its entry is missing, keeps its alias and its +/// dashboard's URL as configured (set, changed or removed after a restart with another +/// `--alias` or `--dashboard-url`) and its /// addresses as the transport gives them, and renews its `alive` stamp now and then /// (see `electorate`). pub(crate) async fn ensure_registered(state: &AppState) { @@ -271,7 +301,15 @@ pub(crate) async fn ensure_registered(state: &AppState) { let local = state.local_id().to_owned(); let own = member_path(&local); let window = state.electorate_window.as_millis() as u64; - let (registered, recorded_alias, recorded_addresses, alive, recorded_window, shortest) = { + let ( + registered, + recorded_alias, + recorded_dashboard, + recorded_addresses, + alive, + recorded_window, + shortest, + ) = { let replica = state.doc.read().await; let registered = store::read_path(&replica.doc, &own) .ok() @@ -280,6 +318,7 @@ pub(crate) async fn ensure_registered(state: &AppState) { ( registered, alias(&replica.doc, &local), + dashboard(&replica.doc, &local), addresses(&replica.doc, &local), millis_at(&replica.doc, &alive_path(&local)), millis_at(&replica.doc, &window_path(&local)), @@ -301,6 +340,9 @@ pub(crate) async fn ensure_registered(state: &AppState) { if let Some(alias) = &wanted { entry["alias"] = json!(alias); } + if let Some(url) = &state.dashboard_url { + entry["dashboard"] = json!(url); + } match state .mutate_doc(move |doc| store::write_path(doc, &own, &entry)) .await @@ -352,6 +394,21 @@ pub(crate) async fn ensure_registered(state: &AppState) { } } + if recorded_dashboard != state.dashboard_url { + let url = state.dashboard_url.clone(); + let local = local.clone(); + match state + .mutate_doc(move |doc| set_dashboard(doc, &local, url.as_deref())) + .await + { + Ok(()) => info!( + url = state.dashboard_url.as_deref(), + "Dashboard URL updated in the member list" + ), + Err(error) => warn!(error = %error, "Failed to record our dashboard URL"), + } + } + if recorded_alias == wanted { return; } diff --git a/src/swarm/peer/worker.rs b/src/swarm/peer/worker.rs index e8e6f8b..6196571 100644 --- a/src/swarm/peer/worker.rs +++ b/src/swarm/peer/worker.rs @@ -170,7 +170,15 @@ pub(crate) async fn run_peer_worker(state: AppState) { crate::swarm::state_file::update(&state, false).await; { let known = state.peer_states.read().await; - state.traffic.retain(|peer| known.contains_key(peer)); + // And their addresses, where the layers below the node count + // (`Traffic::node`). + let addresses: std::collections::HashSet<String> = known + .keys() + .flat_map(|peer| state.transport.identities_of(peer)) + .collect(); + state + .traffic + .retain(|peer| known.contains_key(peer) || addresses.contains(peer)); state .throughput .lock_or_recover() diff --git a/src/swarm/relay.rs b/src/swarm/relay.rs index 1736cec..5ab93f2 100644 --- a/src/swarm/relay.rs +++ b/src/swarm/relay.rs @@ -16,7 +16,10 @@ //! message would announce to every neighbour, and announcements would outnumber the //! data they announce (about 15 to 1 at 250 nodes). //! -//! In a full mesh the writer reaches everyone itself, so nothing is announced. +//! In a full mesh the writer reaches everyone itself, so nothing is announced, but to +//! the neighbours the writer can't reach: those on none of its transports (a member on +//! Tor alone, reached through one on Tor and libp2p, from a writer on libp2p alone). +//! Whom it reaches is what the member list makes its neighbours (`topology::neighbours_of`). use std::collections::{HashMap, HashSet}; use std::sync::atomic::Ordering; @@ -44,7 +47,11 @@ pub(crate) async fn announce(state: &AppState, from: &str) { // While the view is unknown everyone counts as a neighbour, but that isn't a full // mesh: the others may not hear from the writer. if !state.initial_sync_done.load(Ordering::Acquire) - || (topology::view_known(state) && topology::is_full_mesh(state)) + || (topology::view_known(state) + && topology::is_full_mesh(state) + && unreached(state, &HashSet::from([from.to_owned()])) + .await + .is_empty()) { return; } @@ -80,26 +87,55 @@ pub(crate) async fn announce(state: &AppState, from: &str) { /// round with a single sender skips it; with several, each lacks the others' changes, /// so every neighbour is told. async fn announce_now(state: &AppState, senders: &HashSet<String>) { + let neighbours = unreached(state, senders).await; + if neighbours.is_empty() { + return; + } state.stats.announce_round(); + for peer in neighbours { + if let Err(error) = send_status(state, &peer).await { + debug!(peer = %peer, error = %format!("{error:#}"), "Announcing our heads failed"); + } + } +} + +/// The neighbours to tell of the changes `senders` sent us. The peer whose changes we +/// announce has them already, so with a single sender it is skipped; with several, each +/// lacks the others' changes. In a full mesh, also those every sender is a neighbour of: +/// the writers reached them themselves. +async fn unreached(state: &AppState, senders: &HashSet<String>) -> Vec<String> { let skip = match senders.len() { 1 => senders.iter().next().cloned(), _ => None, }; - let neighbours: Vec<String> = state + let full_mesh = topology::view_known(state) && topology::is_full_mesh(state); + // Unknown (a sender not in the member list): nobody counts as reached by it. + let reached: Vec<Option<std::collections::BTreeSet<String>>> = if full_mesh { + senders + .iter() + .map(|sender| topology::neighbours_of(state, sender)) + .collect() + } else { + Vec::new() + }; + let reached_by_all = |peer: &str| { + full_mesh + && reached.iter().zip(senders).all(|(neighbours, sender)| { + sender == peer || neighbours.as_ref().is_some_and(|n| n.contains(peer)) + }) + }; + state .peer_states .read() .await .keys() .filter(|peer| { - skip.as_deref() != Some(peer.as_str()) && topology::is_neighbour(state, peer) + skip.as_deref() != Some(peer.as_str()) + && topology::is_neighbour(state, peer) + && !reached_by_all(peer) }) .cloned() - .collect(); - for peer in neighbours { - if let Err(error) = send_status(state, &peer).await { - debug!(peer = %peer, error = %format!("{error:#}"), "Announcing our heads failed"); - } - } + .collect() } /// Pulls in progress: one neighbour at a time, so each change arrives about once. The diff --git a/src/swarm/topology/mod.rs b/src/swarm/topology/mod.rs index 30fa56e..df75f46 100644 --- a/src/swarm/topology/mod.rs +++ b/src/swarm/topology/mod.rs @@ -133,31 +133,54 @@ pub(crate) fn is_neighbour(state: &AppState, peer: &str) -> bool { pub(crate) fn intended_graph(state: &AppState) -> Option<BTreeMap<String, BTreeSet<String>>> { let view = state.neighbours.read_or_recover(); let view = view.as_ref()?; - let unreachable = |member: &str| view.unreachable.contains(member); - let stable = |member: &str| view.stable.contains(member); let graph = view .members .iter() - .map(|member| { - let neighbours = if member == state.local_id() { - view.neighbours.clone() - } else { - let own = view.transports.get(member).cloned().unwrap_or_default(); - neighbours_across( - &view.transports, - member, - &own, - view.k, - &unreachable, - &stable, - ) - }; - (member.clone(), neighbours) - }) + .map(|member| (member.clone(), view.neighbours_of(state.local_id(), member))) .collect(); Some(graph) } +/// `member`'s neighbours as the member list makes them, as this node sees it (as in +/// `intended_graph`); None until the view is known, or for one the list doesn't have. +pub(crate) fn neighbours_of(state: &AppState, member: &str) -> Option<BTreeSet<String>> { + let view = state.neighbours.read_or_recover(); + let view = view.as_ref()?; + view.members + .iter() + .any(|listed| listed == member) + .then(|| view.neighbours_of(state.local_id(), member)) +} + +/// The transports `member` is on, by the member list (the schemes of its addresses); +/// ours as we are on them. None known: empty. +pub(crate) fn transports_of(state: &AppState, member: &str) -> BTreeSet<String> { + let view = state.neighbours.read_or_recover(); + match view.as_ref() { + Some(view) if member == state.local_id() => view.local_transports.clone(), + Some(view) => view.transports.get(member).cloned().unwrap_or_default(), + None if member == state.local_id() => state.transport.transports().into_iter().collect(), + None => BTreeSet::new(), + } +} + +impl NeighbourView { + fn neighbours_of(&self, local: &str, member: &str) -> BTreeSet<String> { + if member == local { + return self.neighbours.clone(); + } + let own = self.transports.get(member).cloned().unwrap_or_default(); + neighbours_across( + &self.transports, + member, + &own, + self.k, + &|member| self.unreachable.contains(member), + &|member| self.stable.contains(member), + ) + } +} + /// How the neighbours changed. pub(crate) struct NeighbourChange { pub(crate) added: Vec<String>, |