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
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
|
//! 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()));
}
|
