add links relay test

1 files changed, 105 insertions, 0 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;
+    }
+}