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
181
182
183
184
185
186
187
188
189
190
191
192
|
//! Whom neighbours reach (`swarm::reach`): told in STATUSes, as measured, and kept for
//! the session; and the writers' trees built on it (`swarm::tree`), which changes go
//! along however they came.
use std::time::Duration;
use super::Sim;
use crate::state::AppState;
use crate::swarm::bootstrap::Start;
use crate::swarm::{reach, topology, tree};
const SOON: Duration = Duration::from_secs(30);
/// Whether `node` reaches `member`, as `node` measured and told it.
fn reaches(node: &AppState, member: &AppState) -> bool {
reach::published(node).contains(&topology::pair_key(member.local_id()))
}
/// Whether `node` heard from `neighbour` that it reaches `member`.
async fn heard(node: &AppState, neighbour: &AppState, member: &AppState) -> bool {
let Some(peer_state) = node.get_peer_state(neighbour.local_id()).await else {
return false;
};
let said = peer_state.lock().await.reaches.clone();
reach::of(said.as_ref())
.is_some_and(|keys| keys.contains(&topology::pair_key(member.local_id())))
}
/// x, b and c on one net, all members of one full mesh, but nothing passes between x
/// and c: b reaches both and says so; c learns from b that b reaches x, which c itself
/// doesn't.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn neighbours_tell_whom_they_reach() {
let sim = Sim::with_identity(66);
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("b", &["sim"], &[], Start::Genesis);
let c = sim.start_on("c", &["sim"], &["b"], Start::Join);
let x = sim.start_on("x", &["sim"], &["b"], Start::Join);
sim.wait_ready(&[&b, &c, &x], SOON).await;
let (b, c, x) = (&b, &c, &x);
sim.wait_until(
"c heard from b that b reaches x, and c doesn't reach x",
SOON,
move || async move {
reaches(b, x) && reaches(b, c) && !reaches(c, x) && heard(c, b, x).await
},
)
.await;
}
/// The same three: in x's tree (`swarm::tree`), c can't take x as its parent (it
/// doesn't reach it), so it takes b, which comes before it: as far from x, but it
/// reaches x. So x's only child is b, and b's is c.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn writer_trees_go_around_links_down() {
let sim = Sim::with_identity(67);
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("b", &["sim"], &[], Start::Genesis);
let c = sim.start_on("c", &["sim"], &["b"], Start::Join);
let x = sim.start_on("x", &["sim"], &["b"], Start::Join);
sim.wait_ready(&[&b, &c, &x], SOON).await;
let (b, c, x) = (&b, &c, &x);
let writer = topology::pair_key(x.local_id());
sim.wait_until(
"x's tree: b under x, and c under b",
SOON,
move || async move {
let of_x = tree::children(x, writer).await;
let of_b = tree::children(b, writer).await;
heard(c, b, x).await
&& of_x == [b.local_id().to_owned()]
&& of_b == [c.local_id().to_owned()]
},
)
.await;
}
/// What `node` counted: pulls started, changes passed on along writers' trees.
fn counted(node: &AppState) -> (u64, u64) {
let relay = node.stats.snapshot().relay;
(relay.pulls, relay.tree_forwards)
}
/// x, b, c and d, one full mesh, but nothing passes between c and x, nor between c and
/// d: c's parent in x's tree is b, the only member it reaches. And x's sync messages to
/// b are lost, so b gets x's write pulled from d (after d's slow round), not along the
/// tree: pulled, it says nothing of whose; its author does, so b passes it on to c along
/// x's tree all the same, and c pulls nothing.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn pulled_changes_go_along_their_writers_trees() {
// Long, so x's own STATUS (every third keepalive) doesn't make b pull from x, whose
// answers are lost, before d's slow round.
let slow = |config: &mut crate::node::NodeConfig| {
config.timing = crate::state::Timing::new(
Duration::from_secs(5),
super::SYNC_REPLY_TIMEOUT,
crate::transport::sim::DELIVERY_LIMIT,
);
};
let sim = Sim::with_identity(68);
let cut = crate::transport::sim::LinkParams {
loss: 1.0,
..Default::default()
};
for (from, to) in [("x", "c"), ("c", "x"), ("d", "c"), ("c", "d")] {
sim.net.set_link(from, to, cut.clone());
}
sim.net.set_link(
"x",
"b",
crate::transport::sim::LinkParams {
drop: vec![crate::transport::MessageKind::Sync],
..Default::default()
},
);
let b = sim.start_on_with("b", &["sim"], &[], Start::Genesis, slow);
let c = sim.start_on_with("c", &["sim"], &["b"], Start::Join, slow);
let d = sim.start_on_with("d", &["sim"], &["b"], Start::Join, slow);
let x = sim.start_on_with("x", &["sim"], &["b"], Start::Join, slow);
sim.wait_ready(&[&b, &c, &d, &x], SOON).await;
sim.wait_converged(&[&b, &c, &d, &x], SOON * 2).await;
let (b, c, x) = (&b, &c, &x);
let writer = topology::pair_key(x.local_id());
sim.wait_until("c under b in x's tree", SOON, move || async move {
tree::children(b, writer).await == [c.local_id().to_owned()]
})
.await;
let (c_pulls, _) = counted(c);
let (_, b_forwards) = counted(b);
super::write(x, "y", serde_json::json!(1)).await;
sim.wait_until("c has x's write", SOON, move || async move {
super::read(c, "y").await == Some(serde_json::json!(1))
})
.await;
assert!(counted(b).1 > b_forwards, "b passed it on along x's tree");
assert_eq!(counted(c).0, c_pulls, "c pulled nothing");
}
/// x, b and c, nothing between x and c: x's changes reach c through b, along x's tree.
/// x writes every 100 ms for 4 s, many keepalive intervals: the pairs are busy, so no
/// keepalive goes, and no STATUS. b still knows whom c reaches (it said so once, and
/// nothing changed), and passes every write on to c: c pulls none.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn trees_hold_through_continuous_writes() {
let sim = Sim::with_identity(69);
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("b", &["sim"], &[], Start::Genesis);
let c = sim.start_on("c", &["sim"], &["b"], Start::Join);
let x = sim.start_on("x", &["sim"], &["b"], Start::Join);
sim.wait_ready(&[&b, &c, &x], SOON).await;
sim.wait_converged(&[&b, &c, &x], SOON).await;
let (b, c, x) = (&b, &c, &x);
let writer = topology::pair_key(x.local_id());
sim.wait_until("c under b in x's tree", SOON, move || async move {
heard(b, c, b).await && tree::children(b, writer).await == [c.local_id().to_owned()]
})
.await;
let (c_pulls, _) = counted(c);
let (_, b_forwards) = counted(b);
let writes = 40;
for i in 0..writes {
super::write(x, &format!("w/{i}"), serde_json::json!(i)).await;
tokio::time::sleep(Duration::from_millis(100)).await;
}
sim.wait_until("c has x's last write", SOON, move || async move {
super::read(c, &format!("w/{}", writes - 1)).await == Some(serde_json::json!(writes - 1))
})
.await;
// Some may go together, if b had several to pass on at once.
assert!(
counted(b).1 >= b_forwards + writes as u64 / 2,
"b passed the writes on along x's tree"
);
assert_eq!(counted(c).0, c_pulls, "c pulled nothing");
}
|
