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
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
|
//! Changes passed on in a full mesh (`swarm::relay`) to the members the writer can't
//! reach: those on none of its transports, or behind a link that fails; and the links
//! a pull makes eager.
use std::time::Duration;
use serde_json::json;
use super::{SYNC_REPLY_TIMEOUT, Sim, read, write};
use crate::state::Timing;
use crate::swarm::bootstrap::Start;
use crate::swarm::topology;
use crate::transport::sim::DELIVERY_LIMIT;
const SOON: Duration = Duration::from_secs(30);
/// a on `sim` alone, c on `sim2` alone, b on both; a full mesh at each (every member
/// it reaches is its neighbour), so a pushes its write to b alone, and b, which only
/// received it, must announce it to c. With a long keepalive, the regular STATUS (every
/// third keepalive, 15 s) can't be what brings it: c has it within a second (an
/// announcement waits 200 ms, then c pulls). Without the announcement, a STATUS that
/// happens to fall into that second would pass it too, once in about 15 runs.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn changes_reach_members_the_writer_cant_reach_at_once() {
let keepalive = Duration::from_secs(5);
let slow = move |config: &mut crate::node::NodeConfig| {
config.timing = Timing::new(keepalive, SYNC_REPLY_TIMEOUT, DELIVERY_LIMIT);
};
let sim = Sim::with_identity(62);
let a = sim.start_on_with("a", &["sim"], &[], Start::Genesis, slow);
let b = sim.start_on_with("b", &["sim", "sim2"], &["a"], Start::Join, slow);
let c = sim.start_on_with("c", &["sim2"], &["b"], Start::Join, slow);
sim.wait_ready(&[&a, &b, &c], SOON).await;
let c_id = c.local_id().to_owned();
{
let (a, b, c_id) = (&a, &b, c_id.clone());
sim.wait_until(
"a and b know their neighbours, a full mesh, c beyond a's reach",
SOON,
move || {
let c_id = c_id.clone();
async move {
topology::view_known(a)
&& topology::view_known(b)
&& topology::is_full_mesh(a)
&& topology::is_full_mesh(b)
&& topology::is_neighbour(b, &c_id)
&& topology::transports_of(a, &c_id).contains("sim2")
&& !a.transport.can_reach(&c_id)
}
},
)
.await;
}
sim.wait_converged(&[&a, &b, &c], SOON).await;
write(&a, "x", json!(1)).await;
{
let c = &c;
sim.wait_until(
"c has a's write within a second",
Duration::from_secs(1),
move || async move { read(c, "x").await == Some(json!(1)) },
)
.await;
}
}
/// One transport, a full mesh by the member list, but the writer x reaches b alone (as
/// one behind NAT may): what x writes reaches c through b, at once, though the member
/// list makes c x's neighbour too.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn changes_reach_members_the_writer_fails_to_reach_at_once() {
let keepalive = Duration::from_secs(5);
let slow = move |config: &mut crate::node::NodeConfig| {
config.timing = Timing::new(keepalive, SYNC_REPLY_TIMEOUT, DELIVERY_LIMIT);
};
let sim = Sim::with_identity(63);
// Nothing between x and c, either way.
let cut = crate::transport::sim::LinkParams {
loss: 1.0,
..Default::default()
};
sim.net.set_link("x", "c", cut.clone());
sim.net.set_link("c", "x", cut);
let b = sim.start_on_with("b", &["sim"], &[], Start::Genesis, slow);
let c = sim.start_on_with("c", &["sim"], &["b"], Start::Join, slow);
let x = sim.start_on_with("x", &["sim"], &["b"], Start::Join, slow);
sim.wait_ready(&[&b, &c, &x], SOON).await;
let (c_id, x_id) = (c.local_id().to_owned(), x.local_id().to_owned());
{
let (b, c_id, x_id) = (&b, c_id.clone(), x_id.clone());
sim.wait_until(
"b knows its neighbours, a full mesh, x and c among them",
SOON,
move || {
let (c_id, x_id) = (c_id.clone(), x_id.clone());
async move {
topology::view_known(b)
&& topology::is_full_mesh(b)
&& topology::is_neighbour(b, &c_id)
&& topology::is_neighbour(b, &x_id)
}
},
)
.await;
}
sim.wait_converged(&[&b, &c, &x], SOON).await;
write(&x, "y", json!(1)).await;
{
let c = &c;
sim.wait_until(
"c has x's write within a second",
Duration::from_secs(1),
move || async move { read(c, "y").await == Some(json!(1)) },
)
.await;
}
}
/// Whether `node` asked `peer` for its changes eagerly, and whether `peer` heard so.
async fn eager(node: &crate::state::AppState, peer: &crate::state::AppState) -> bool {
let asked = match node.get_peer_state(peer.local_id()).await {
Some(state) => state.lock().await.eager_from,
None => false,
};
let heard = match peer.get_peer_state(node.local_id()).await {
Some(state) => state.lock().await.eager_to,
None => false,
};
asked && heard
}
/// As above, x reaching b alone: c pulls x's first write from b, which makes the link
/// from b to c eager; x's next write b forwards to c as it applies it, without
/// announcing it. The link from x's side stays lazy: nothing was pulled over it.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn pulls_make_links_eager() {
let keepalive = Duration::from_secs(5);
let slow = move |config: &mut crate::node::NodeConfig| {
config.timing = Timing::new(keepalive, SYNC_REPLY_TIMEOUT, DELIVERY_LIMIT);
};
let sim = Sim::with_identity(65);
let cut = crate::transport::sim::LinkParams {
loss: 1.0,
..Default::default()
};
sim.net.set_link("x", "c", cut.clone());
sim.net.set_link("c", "x", cut);
let b = sim.start_on_with("b", &["sim"], &[], Start::Genesis, slow);
let c = sim.start_on_with("c", &["sim"], &["b"], Start::Join, slow);
let x = sim.start_on_with("x", &["sim"], &["b"], Start::Join, slow);
sim.wait_ready(&[&b, &c, &x], SOON).await;
// c may have pulled already: what x writes when it registers comes that way too.
sim.wait_converged(&[&b, &c, &x], SOON).await;
write(&x, "y", json!(1)).await;
{
let (b, c) = (&b, &c);
sim.wait_until(
"c pulled x's write from b, and the link from b is eager",
SOON,
move || async move { read(c, "y").await == Some(json!(1)) && eager(c, b).await },
)
.await;
}
assert!(!eager(&b, &x).await, "nothing was pulled from x");
write(&x, "y", json!(2)).await;
{
let c = &c;
sim.wait_until(
"c has x's next write within a second",
Duration::from_secs(1),
move || async move { read(c, "y").await == Some(json!(2)) },
)
.await;
}
}
|
