add links relay test

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))]