diff options
| -rw-r--r-- | docs/development.md | 5 | ||||
| -rw-r--r-- | src/simtest/links.rs | 105 | ||||
| -rw-r--r-- | src/simtest/mod.rs | 62 | ||||
| -rw-r--r-- | src/swarm/links.rs | 33 |
4 files changed, 204 insertions, 1 deletions
diff --git a/docs/development.md b/docs/development.md index 93726b1..d7f6068 100644 --- a/docs/development.md +++ b/docs/development.md @@ -50,3 +50,8 @@ take as long as the protocol's timeouts make them, however fast the machine, and busy machine some fail now and then (a node's timer fires before the work it waits for is done). One failing alone (`cargo sim <name>`) and passing then is that, not a regression; one failing again and again is a regression. + +A node on several transports is a node on several simulated nets, one per scheme +(`Sim::start_on`, under the identity layer): nodes on different nets can't reach each +other, as one on Tor alone and one on libp2p alone can't, so a swarm whose transports +don't all meet is simulated with the simulated transport alone. 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))] |