swarm: relay links messages

2 files changed, 434 insertions, 24 deletions

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 {