diff options
Diffstat (limited to 'src')
| -rw-r--r-- | src/simtest/links.rs | 105 | ||||
| -rw-r--r-- | src/simtest/mod.rs | 62 | ||||
| -rw-r--r-- | src/swarm/links.rs | 33 |
3 files changed, 199 insertions, 1 deletions
diff --git a/src/simtest/links.rs b/src/simtest/links.rs new file mode 100644 index 0000000..44836bd --- /dev/null +++ b/src/simtest/links.rs @@ -0,0 +1,105 @@ +//! Links relayed (`swarm::links`): a member the asker can't reach, on a transport the +//! asker isn't on, is asked through a node on both, for as long as the asker renews. + +use std::time::Duration; + +use serde_json::json; + +use super::{Sim, read, write}; +use crate::state::AppState; +use crate::swarm::bootstrap::Start; +use crate::swarm::{links, members, topology}; + +const SOON: Duration = Duration::from_secs(20); + +/// Whether `node` has `member`'s links, listing `peer`. +fn has_links(node: &AppState, member: &str, peer: &str) -> bool { + node.links + .replies() + .iter() + .any(|(from, _, links)| from == member && links.iter().any(|link| link.peer == peer)) +} + +/// a on `sim` alone, c on `sim2` alone, b on both: a and c never reach each other, yet +/// a's writes reach c, and a gets c's links through b, while it renews; once it stops, +/// the subscription runs out at b and at c. +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn links_are_relayed_to_members_out_of_reach() { + let sim = Sim::with_identity(61); + let a = sim.start_on("a", &["sim"], &[], Start::Genesis); + let b = sim.start_on("b", &["sim", "sim2"], &["a"], Start::Join); + let c = sim.start_on("c", &["sim2"], &["b"], Start::Join); + sim.wait_ready(&[&a, &b, &c], SOON).await; + let (a_id, b_id, c_id) = ( + a.local_id().to_owned(), + b.local_id().to_owned(), + c.local_id().to_owned(), + ); + + { + let (a, c_id) = (&a, c_id.clone()); + sim.wait_until("a knows c, out of its reach", SOON, move || { + let c_id = c_id.clone(); + async move { + let listed = members::members(&a.doc.read().await.doc).contains(&c_id); + // On `sim2` as a's view of the member list has it, which picks the relay. + let on = topology::transports_of(a, &c_id); + listed && on.contains("sim2") && !a.transport.can_reach(&c_id) + } + }) + .await; + } + + // Changes pass through b. + write(&a, "x", json!(1)).await; + { + let c = &c; + sim.wait_until("c has a's write", SOON, move || async move { + read(c, "x").await == Some(json!(1)) + }) + .await; + } + + // c's links reach a through b; b asks c for a. + let lease = Duration::from_secs(3); + links::ask_all(&a, vec![b_id.clone(), c_id.clone()], lease).await; + { + let (a, b_id, c_id) = (&a, b_id.clone(), c_id.clone()); + sim.wait_until("a has c's links, relayed by b", SOON, move || { + let (b_id, c_id) = (b_id.clone(), c_id.clone()); + async move { has_links(a, &c_id, &b_id) } + }) + .await; + } + assert!( + b.links + .relaying() + .contains(&(c_id.clone(), vec![a_id.clone()])), + "{:?}", + b.links.relaying() + ); + assert!(c.links.subscribers().contains(&b_id)); + assert!(!c.links.subscribers().contains(&a_id)); + + // Renewed: it lasts beyond its lease. + for _ in 0..3 { + tokio::time::sleep(lease / 2).await; + links::ask_all(&a, vec![c_id.clone()], lease).await; + } + assert!(!b.links.relaying().is_empty()); + assert!(c.links.subscribers().contains(&b_id)); + + // Not renewed: it runs out at every hop, without a message. + { + let (b, c, b_id) = (&b, &c, b_id.clone()); + sim.wait_until( + "the relayed subscription ran out at b and at c", + lease * 2, + move || { + let b_id = b_id.clone(); + async move { b.links.relaying().is_empty() && !c.links.subscribers().contains(&b_id) } + }, + ) + .await; + } +} diff --git a/src/simtest/mod.rs b/src/simtest/mod.rs index a79bd2b..76583de 100644 --- a/src/simtest/mod.rs +++ b/src/simtest/mod.rs @@ -11,6 +11,7 @@ mod deltas; mod identity; +mod links; mod remakes; mod soak; mod timing; @@ -47,7 +48,14 @@ pub(crate) const SOON: Duration = Duration::from_secs(10); /// A simulated swarm: the net and the nodes started on it. pub(crate) struct Sim { + /// The net of the transport `sim`, every node's but those started on others + /// (`start_on`). pub(crate) net: SimNet, + /// Further nets, by the scheme of their transport: nodes on different nets can't + /// reach each other, as nodes on different transports (one on Tor alone, one on + /// libp2p alone) can't. Made when first used. + nets: std::sync::Mutex<std::collections::HashMap<&'static str, SimNet>>, + seed: u64, /// Nodes under the identity layer (`transport::identity`), named by node IDs; else /// by their names on the net. identity: bool, @@ -60,6 +68,8 @@ impl Sim { init_logging(); Self { net: SimNet::new(seed), + nets: Default::default(), + seed, identity: false, nodes: Default::default(), } @@ -99,8 +109,58 @@ impl Sim { start: Start, adjust: impl FnOnce(&mut NodeConfig), ) -> AppState { + self.start_node(id, self.net.transport(id), seeds, start, adjust) + } + + /// [`start`](Self::start) under the identity layer, on the nets of `schemes` + /// instead of `sim`'s alone (the first preferred): it reaches only nodes on one of + /// them. Seeds by name are resolved on the first. + pub(crate) fn start_on( + &self, + id: &str, + schemes: &[&'static str], + seeds: &[&str], + start: Start, + ) -> AppState { + assert!( + self.identity, + "transports by scheme need the identity layer" + ); + let mut transports = schemes.iter().map(|scheme| { + let transport: Arc<dyn crate::transport::Transport> = + self.net_for(scheme).transport(id); + (*scheme, transport) + }); + let (scheme, first) = transports.next().expect("a transport at least"); + let more: Vec<_> = transports.collect(); + let key = Arc::new(sim_key(id)); + self.start_node(id, first, seeds, start, move |config| { + config.identity = Some(IdentityConfig { key, scheme, more }); + }) + } + + /// The net of the transport `scheme`. + pub(crate) fn net_for(&self, scheme: &'static str) -> SimNet { + if scheme == "sim" { + return self.net.clone(); + } + let mut nets = self.nets.lock().unwrap(); + let index = nets.len() as u64 + 1; + nets.entry(scheme) + .or_insert_with(|| SimNet::new(self.seed.wrapping_add(index))) + .clone() + } + + fn start_node( + &self, + id: &str, + transport: Arc<dyn crate::transport::Transport>, + seeds: &[&str], + start: Start, + adjust: impl FnOnce(&mut NodeConfig), + ) -> AppState { let mut config = NodeConfig { - transport: self.net.transport(id), + transport, identity: self.identity.then(|| IdentityConfig { key: Arc::new(sim_key(id)), scheme: "sim", diff --git a/src/swarm/links.rs b/src/swarm/links.rs index 80def7a..cdbae67 100644 --- a/src/swarm/links.rs +++ b/src/swarm/links.rs @@ -131,6 +131,39 @@ impl Links { .collect() } + /// 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))] |