diff options
Diffstat (limited to 'src/simtest/links.rs')
| -rw-r--r-- | src/simtest/links.rs | 259 |
1 files changed, 0 insertions, 259 deletions
diff --git a/src/simtest/links.rs b/src/simtest/links.rs deleted file mode 100644 index 652e3ed..0000000 --- a/src/simtest/links.rs +++ /dev/null @@ -1,259 +0,0 @@ -//! 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; - } -} - -/// a, b and c all on `sim`, but nothing passes between a and c (as behind NAT): once a -/// judges c unreachable, it asks c's links through b, though it has c's address. -#[tokio::test(flavor = "multi_thread", worker_threads = 4)] -async fn links_are_relayed_to_members_that_dont_answer() { - let sim = Sim::with_identity(64); - let cut = crate::transport::sim::LinkParams { - loss: 1.0, - ..Default::default() - }; - sim.net.set_link("a", "c", cut.clone()); - sim.net.set_link("c", "a", cut); - let b = sim.start_on("b", &["sim"], &[], Start::Genesis); - let a = sim.start_on("a", &["sim"], &["b"], Start::Join); - let c = sim.start_on("c", &["sim"], &["b"], Start::Join); - sim.wait_ready(&[&a, &b, &c], SOON).await; - let (b_id, c_id) = (b.local_id().to_owned(), c.local_id().to_owned()); - { - let (a, c_id) = (&a, c_id.clone()); - sim.wait_until( - "a knows c, and that it doesn't answer", - SOON * 3, - move || { - let c_id = c_id.clone(); - async move { - let listed = members::members(&a.doc.read().await.doc).contains(&c_id); - listed && a.transport.can_reach(&c_id) && !topology::reaches(a, &c_id).await - } - }, - ) - .await; - } - links::ask_all( - &a, - vec![b_id.clone(), c_id.clone()], - Duration::from_secs(10), - ) - .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!(!c.links.subscribers().contains(&a.local_id().to_owned())); -} - -/// a and c can't talk; b and d both could relay. Once the relay a asks through is cut -/// off from c (and from the other relay, so it can't pass the ask on), it brings -/// nothing: after `FAILED_AFTER` a's renewal goes through the other one, and the -/// subscription at the first runs out by itself. -#[tokio::test(flavor = "multi_thread", worker_threads = 4)] -async fn links_fail_over_to_another_relay() { - let sim = Sim::with_identity(67); - let cut = crate::transport::sim::LinkParams { - loss: 1.0, - ..Default::default() - }; - let cut_between = |x: &str, y: &str| { - sim.net.set_link(x, y, cut.clone()); - sim.net.set_link(y, x, cut.clone()); - }; - cut_between("a", "c"); - let b = sim.start_on("b", &["sim"], &[], Start::Genesis); - let d = sim.start_on("d", &["sim"], &["b"], Start::Join); - let a = sim.start_on("a", &["sim"], &["b"], Start::Join); - let c = sim.start_on("c", &["sim"], &["b"], Start::Join); - sim.wait_ready(&[&a, &b, &c, &d], SOON).await; - let (a_id, c_id) = (a.local_id().to_owned(), c.local_id().to_owned()); - { - let (a, c_id) = (&a, c_id.clone()); - sim.wait_until( - "a knows c, and that it doesn't answer", - SOON * 3, - move || { - let c_id = c_id.clone(); - async move { - let listed = members::members(&a.doc.read().await.doc).contains(&c_id); - listed && !topology::reaches(a, &c_id).await - } - }, - ) - .await; - } - - let lease = Duration::from_secs(3); - links::ask_all(&a, vec![c_id.clone()], lease).await; - { - let (b, d, c_id, a_id) = (&b, &d, c_id.clone(), a_id.clone()); - sim.wait_until("one of b and d relays c's links to a", SOON, move || { - let relayed = (c_id.clone(), vec![a_id.clone()]); - async move { - b.links.relaying().contains(&relayed) || d.links.relaying().contains(&relayed) - } - }) - .await; - } - let relayed = (c_id.clone(), vec![a_id.clone()]); - // With their names on the simulated net, which links are cut by. - let ((first, first_name), (other, other_name)) = if b.links.relaying().contains(&relayed) { - ((&b, "b"), (&d, "d")) - } else { - ((&d, "d"), (&b, "b")) - }; - - // The relay in use reaches c no more, nor the other relay; a still reaches it. - cut_between(first_name, "c"); - cut_between(first_name, other_name); - let cut_at = tokio::time::Instant::now(); - - // a renews as a dashboard does, and gets c's links through the other relay. - let renewing = { - let (a, c_id) = (a.clone(), c_id.clone()); - tokio::spawn(async move { - loop { - tokio::time::sleep(lease / 2).await; - links::ask_all(&a, vec![c_id.clone()], lease).await; - } - }) - }; - { - let (a, other, relayed, c_id) = (&a, other, relayed.clone(), c_id.clone()); - sim.wait_until( - "the other relay relays c's links to a, and they arrive", - SOON * 3, - move || { - let (relayed, c_id) = (relayed.clone(), c_id.clone()); - async move { - other.links.relaying().contains(&relayed) - && a.links - .replies() - .iter() - .any(|(from, age, _)| from == &c_id && *age < cut_at.elapsed()) - } - }, - ) - .await; - } - { - let (first, relayed) = (first, relayed.clone()); - sim.wait_until( - "the first relay's subscription ran out", - lease * 2, - move || { - let relayed = relayed.clone(); - async move { !first.links.relaying().contains(&relayed) } - }, - ) - .await; - } - renewing.abort(); -} |