diff options
Diffstat (limited to 'src/swarm/links.rs')
| -rw-r--r-- | src/swarm/links.rs | 585 |
1 files changed, 556 insertions, 29 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(), |