diff options
Diffstat (limited to 'src/simtest')
| -rw-r--r-- | src/simtest/links.rs | 105 | ||||
| -rw-r--r-- | src/simtest/mod.rs | 78 | ||||
| -rw-r--r-- | src/simtest/relay.rs | 66 |
3 files changed, 248 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 61c81d1..267768f 100644 --- a/src/simtest/mod.rs +++ b/src/simtest/mod.rs @@ -11,6 +11,8 @@ mod deltas; mod identity; +mod links; +mod relay; mod remakes; mod soak; mod timing; @@ -47,7 +49,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 +69,8 @@ impl Sim { init_logging(); Self { net: SimNet::new(seed), + nets: Default::default(), + seed, identity: false, nodes: Default::default(), } @@ -99,8 +110,71 @@ 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 { + self.start_on_with(id, schemes, seeds, start, |_| {}) + } + + /// [`start_on`](Self::start_on), with the configuration adjusted first. + pub(crate) fn start_on_with( + &self, + id: &str, + schemes: &[&'static str], + seeds: &[&str], + start: Start, + adjust: impl FnOnce(&mut NodeConfig), + ) -> 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 }); + adjust(config); + }) + } + + /// 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", @@ -112,8 +186,10 @@ impl Sim { timing: Timing::new(KEEPALIVE, SYNC_REPLY_TIMEOUT, DELIVERY_LIMIT), neighbours_k: 8, alias: Some(id.to_owned()), + dashboard_url: None, electorate_window: Duration::from_secs(30 * 24 * 3600), clock_skew_warn: Duration::from_secs(2), + stream_heartbeat: Duration::from_secs(30), superseded_writes: Default::default(), blob_chunk_bytes: 64 * 1024, swarm_key: None, diff --git a/src/simtest/relay.rs b/src/simtest/relay.rs new file mode 100644 index 0000000..9b016b8 --- /dev/null +++ b/src/simtest/relay.rs @@ -0,0 +1,66 @@ +//! Changes passed on in a full mesh (`swarm::relay`) to the members the writer can't +//! reach: those on none of its transports. + +use std::time::Duration; + +use serde_json::json; + +use super::{SYNC_REPLY_TIMEOUT, Sim, read, write}; +use crate::state::Timing; +use crate::swarm::bootstrap::Start; +use crate::swarm::topology; +use crate::transport::sim::DELIVERY_LIMIT; + +const SOON: Duration = Duration::from_secs(30); + +/// a on `sim` alone, c on `sim2` alone, b on both; a full mesh at each (every member +/// it reaches is its neighbour), so a pushes its write to b alone, and b, which only +/// received it, must announce it to c. With a long keepalive, the regular STATUS (every +/// third keepalive, 15 s) can't be what brings it: c has it within a second (an +/// announcement waits 200 ms, then c pulls). Without the announcement, a STATUS that +/// happens to fall into that second would pass it too, once in about 15 runs. +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn changes_reach_members_the_writer_cant_reach_at_once() { + let keepalive = Duration::from_secs(5); + let slow = move |config: &mut crate::node::NodeConfig| { + config.timing = Timing::new(keepalive, SYNC_REPLY_TIMEOUT, DELIVERY_LIMIT); + }; + let sim = Sim::with_identity(62); + let a = sim.start_on_with("a", &["sim"], &[], Start::Genesis, slow); + let b = sim.start_on_with("b", &["sim", "sim2"], &["a"], Start::Join, slow); + let c = sim.start_on_with("c", &["sim2"], &["b"], Start::Join, slow); + sim.wait_ready(&[&a, &b, &c], SOON).await; + let c_id = c.local_id().to_owned(); + { + let (a, b, c_id) = (&a, &b, c_id.clone()); + sim.wait_until( + "a and b know their neighbours, a full mesh, c beyond a's reach", + SOON, + move || { + let c_id = c_id.clone(); + async move { + topology::view_known(a) + && topology::view_known(b) + && topology::is_full_mesh(a) + && topology::is_full_mesh(b) + && topology::is_neighbour(b, &c_id) + && topology::transports_of(a, &c_id).contains("sim2") + && !a.transport.can_reach(&c_id) + } + }, + ) + .await; + } + sim.wait_converged(&[&a, &b, &c], SOON).await; + + write(&a, "x", json!(1)).await; + { + let c = &c; + sim.wait_until( + "c has a's write within a second", + Duration::from_secs(1), + move || async move { read(c, "x").await == Some(json!(1)) }, + ) + .await; + } +} |