1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
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;
}
}
|
