diff options
| -rw-r--r-- | docs/architecture.md | 2 | ||||
| -rw-r--r-- | docs/dashboard.md | 23 | ||||
| -rw-r--r-- | src/api/dashboard/mod.rs | 9 | ||||
| -rw-r--r-- | src/swarm/links.rs | 446 | ||||
| -rw-r--r-- | src/swarm/topology/mod.rs | 12 | ||||
| -rw-r--r-- | src/transport/mod.rs | 23 | ||||
| -rw-r--r-- | src/transport/wire.rs | 10 |
7 files changed, 490 insertions, 35 deletions
diff --git a/docs/architecture.md b/docs/architecture.md index 07edbfb..9b9db71 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -233,7 +233,7 @@ replies, admit (swarm key), mark the peer seen, record its epoch (ignoring late then by type: HELLO answered, HELLO_REPLY establishes membership and discovers peers, sync queued on its lane if on our epoch (held back from everyone but the source while catching up), STATUS answered with a STATUS_REPLY and compared for divergence, blobs served or collected, -LINKS answered with the node's links if the asker is a member, and again on each change while it subscribes (`swarm::links`, for the dashboard); +LINKS answered with the node's links if the asker is a member, and again on each change while it subscribes, or with other members' links it asks for the asker, which can't reach them (`swarm::links`, for the dashboard); PING and PONG (the paths layer answered the PING already) only mark the peer seen, and after a PONG the peer's round-trip time is its path's again. diff --git a/docs/dashboard.md b/docs/dashboard.md index a93e4d4..c614aab 100644 --- a/docs/dashboard.md +++ b/docs/dashboard.md @@ -323,14 +323,33 @@ older than 90 s is left out. A member keeps 64 subscribers at most. With `links: and the members are only asked every 30 s, for every page open on this node, until asked otherwise or the node restarts. +A member this node can't reach (on none of its transports: one on Tor alone, from a +node on libp2p alone) is asked through a peer that can: an active member it reaches +that, by the member list, shares a transport with the member, else one on a transport +this node isn't on (the lowest pair score of several). The LINKS names the members to +ask (`of`): the peer asks them on this node's behalf, itself or again through a peer +of its own (three nodes at most), for as long as this node asked, and passes their +links on in its replies (`relayed`, each with its age), several together: it holds +what arrives a while. The asker's window (250 ms, at most 5 s) is for the whole way: +each hop holds its share of what is left, the window divided among it and the hops +that may follow (at least 10 ms), and passes the rest on, so links reach the asker +within the window however long the chain. Such a subscription isn't ended by a message: every hop's runs +out unless renewed, so it ends within 90 s of the page closing or the node going away. +A hop renews its own as soon as it is renewed, for the same time, so the hops run in +step, each a message's delay behind the one before (it never runs out first), and a +renewal keeps its nonce, so what is on its way under the last one still counts. A +renewal lost on the way leaves the hops behind it to run out at the old time: with 90 s +renewed every 30 s, one lost costs nothing. + The links a member reports: its neighbours and the peers it holds dormant (a neighbour that goes dormant is one no more, but stays listed until it is heard from again), by which transport, whether only through a relay, the round trip (as the paths layer measures it), whether it holds the peer dormant or the path is down, the peer's other transports with a path up (each with its fastest path's round trip, and whether it goes only through a relay), and the peer's clock against the member's ([`measurements.md`](measurements.md), -Per peer, `clock`). A member answers a LINKS from members only, at most every 5 s, and -every build answers, with or without the feature (`swarm::links`). +Per peer, `clock`). A member answers a LINKS from members only, at most every 5 s (one +asking it to relay, each time), and every build answers, with or without the feature +(`swarm::links`). **Clocks.** The clocks in the links chain every member's clock to this node's, along the narrowest bounds, and the map takes each against their median: a member more than diff --git a/src/api/dashboard/mod.rs b/src/api/dashboard/mod.rs index 14321a8..8d954c0 100644 --- a/src/api/dashboard/mod.rs +++ b/src/api/dashboard/mod.rs @@ -302,14 +302,13 @@ impl Graph { let replica = state.doc.read().await; members::members(&replica.doc) }; - for member in members + let members: Vec<String> = members .into_iter() .filter(|member| member != state.local_id()) .take(MAX_ASKED) - { - let state = state.clone(); - tokio::spawn(async move { links::ask(&state, &member, lease).await }); - } + .collect(); + // Those we can't reach, through peers that can (`links::ask_all`). + links::ask_all(state, members, lease).await; } } diff --git a/src/swarm/links.rs b/src/swarm/links.rs index 1d5436e..80def7a 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,15 @@ 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; +/// 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); +/// 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 +90,26 @@ 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>>, +} + +/// 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 +131,24 @@ impl Links { .collect() } - /// The members our subscriptions are with (not run out). + /// 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(); - self.inner - .lock_or_recover() + 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() } } @@ -222,18 +271,156 @@ pub(crate) async fn ask(state: &AppState, member: &str, subscribe: Duration) { 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. 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 members: Vec<String> = members.into_iter().take(MAX_RELAYED).collect(); + 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(); + let request = ParsedMessage::Links(LinksRequest { + epoch, + epoch_seq, + nonce, + subscribe_secs: u32::try_from(subscribe.as_secs()).unwrap_or(u32::MAX), + of: members, + 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, @@ -245,6 +432,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"); @@ -263,6 +453,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(); @@ -298,33 +492,236 @@ 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 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(); + 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)).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; + } } - reply.links.truncate(MAX_LINKS); - for link in &mut reply.links { - link.others.truncate(MAX_OTHERS); + 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 chunk in relayed.chunks(MAX_RELAYED) { + let reply = ParsedMessage::LinksReply(LinksReply { + epoch, + epoch_seq, + nonce, + links: Vec::new(), + relayed: chunk.to_vec(), + }); + if let Err(error) = state.transport.send_message(asker, &reply).await { + debug!(asker = %asker, error = %format!("{error:#}"), "Relayed LINKS_REPLY failed"); + } + } + } +} + +/// 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, + ); + } + } + } } - if inner.replies.len() >= MAX_KEPT && !inner.replies.contains_key(peer) { - inner.replies.clear(); + 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 @@ -374,6 +771,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"); diff --git a/src/swarm/topology/mod.rs b/src/swarm/topology/mod.rs index 946fc33..df75f46 100644 --- a/src/swarm/topology/mod.rs +++ b/src/swarm/topology/mod.rs @@ -152,6 +152,18 @@ pub(crate) fn neighbours_of(state: &AppState, member: &str) -> Option<BTreeSet<S .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 { diff --git a/src/transport/mod.rs b/src/transport/mod.rs index 96f7651..b8b57af 100644 --- a/src/transport/mod.rs +++ b/src/transport/mod.rs @@ -409,8 +409,16 @@ pub(crate) struct LinksRequest { pub(crate) nonce: u64, /// For this long (seconds), a LINKS_REPLY again whenever the links change (a /// subscription; the next LINKS renews it). 0: this reply only, and a subscription - /// ends. + /// ends (a relayed one runs out instead). pub(crate) subscribe_secs: u32, + /// Members the receiver is to ask on the sender's behalf, which the sender can't + /// reach, passing their links on (relayed, `swarm::links`); none: the receiver's own. + pub(crate) of: Vec<String>, + /// How long relayed links may be held to send several in one LINKS_REPLY, by the + /// receiver and the hops after it together (milliseconds; `swarm::links`). + pub(crate) batch_ms: u32, + /// How many more nodes a relayed request may go through, after the receiver. + pub(crate) hops: u8, } /// LINKS_REPLY: how the sender is linked to its neighbours, as it measures them. @@ -422,7 +430,18 @@ pub(crate) struct LinksReply { pub(crate) epoch_seq: u32, /// The request's. pub(crate) nonce: u64, - /// At most `swarm::links::MAX_LINKS`. + /// The sender's own, if asked for; at most `swarm::links::MAX_LINKS`. + pub(crate) links: Vec<Link>, + /// Other members' links, passed on (`LinksRequest::of`). + pub(crate) relayed: Vec<RelayedLinks>, +} + +/// A member's links as a LINKS_REPLY passes them on. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub(crate) struct RelayedLinks { + pub(crate) member: String, + /// How long ago the member reported them (milliseconds). + pub(crate) age_ms: u32, pub(crate) links: Vec<Link>, } diff --git a/src/transport/wire.rs b/src/transport/wire.rs index 7717200..0769e9a 100644 --- a/src/transport/wire.rs +++ b/src/transport/wire.rs @@ -14,7 +14,7 @@ use super::{EncodedMessage, MessageKind, ParsedMessage}; /// The wire format's version, the first byte of every payload. Bump it with every change /// to a message type or a codec (`golden_bytes` fails then). -pub(crate) const VERSION: u8 = 14; +pub(crate) const VERSION: u8 = 15; /// What the second byte of a payload says. fn tag(kind: MessageKind) -> u8 { @@ -428,6 +428,9 @@ pub(crate) fn samples() -> Vec<ParsedMessage> { epoch_seq: 1, nonce: 3, subscribe_secs: 60, + of: vec!["q".into()], + batch_ms: 250, + hops: 2, }), ParsedMessage::LinksReply(crate::transport::LinksReply { epoch, @@ -450,6 +453,11 @@ pub(crate) fn samples() -> Vec<ParsedMessage> { within_ms: 40, }), }], + relayed: vec![crate::transport::RelayedLinks { + member: "q".into(), + age_ms: 1200, + links: Vec::new(), + }], }), ] } |