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
|
//! Many nodes under realistic conditions for a while. Ignored by default (a few
//! minutes in a debug build): `cargo test soak -- --ignored --nocapture`.
use std::time::Duration;
use serde_json::json;
use super::{Sim, write};
use crate::state::AppState;
use crate::swarm::bootstrap::Start;
use crate::swarm::members;
use crate::swarm::stats::Propagation;
use crate::transport::sim::LinkParams;
const NODES: usize = 25;
/// One per second, round robin over the nodes.
const WRITES: usize = 60;
/// 99% of changes reach a node within this, from the writer's commit. Measured: p99
/// within the 2.5 s bucket, mean about 0.7 s (October 2026); the bound is the next
/// bucket up.
const P99_BOUND_MS: u64 = 5_000;
/// 25 nodes (a partial mesh: 8 neighbours each) over links with 2% loss, 1%
/// duplication and 50–500 ms delay; one write per second for a minute. All converge,
/// and 99% of changes reach each node within `P99_BOUND_MS`.
#[tokio::test(flavor = "multi_thread", worker_threads = 8)]
#[ignore]
async fn soak_25_nodes() {
let sim = Sim::new(25);
sim.net.set_default_link(LinkParams {
loss: 0.02,
duplicate: 0.01,
delay: (Duration::from_millis(50), Duration::from_millis(500)),
..LinkParams::default()
});
let ids: Vec<String> = (0..NODES).map(|i| format!("n{i:02}")).collect();
let mut nodes = vec![sim.start(&ids[0], &[], Start::Genesis)];
for id in &ids[1..] {
nodes.push(sim.start(id, &[ids[0].as_str()], Start::Join));
}
let all: Vec<&AppState> = nodes.iter().collect();
let all = all.as_slice();
sim.wait_ready(all, Duration::from_secs(120)).await;
sim.wait_until(
"every node knows every member",
Duration::from_secs(60),
move || async move {
for node in all {
if members::members(&node.doc.read().await.doc).len() < NODES {
return false;
}
}
true
},
)
.await;
for i in 0..WRITES {
write(&nodes[i % NODES], &format!("soak/{i}"), json!(i)).await;
tokio::time::sleep(Duration::from_secs(1)).await;
}
sim.wait_converged(all, Duration::from_secs(60)).await;
let mut total = Propagation::default();
for node in all {
let propagation = node.stats.snapshot().propagation;
for (sum, count) in total.buckets.iter_mut().zip(propagation.buckets) {
*sum += count;
}
total.count += propagation.count;
total.sum_ms += propagation.sum_ms;
}
let p99 = percentile_bound(&total, 0.99);
println!(
"propagation: {} changes, mean {} ms, p99 within {}, buckets {:?}; net {:?}",
total.count,
total.sum_ms / total.count.max(1),
p99.map_or("more than an hour".to_owned(), |ms| format!("{ms} ms")),
total.buckets().collect::<Vec<_>>(),
sim.net.counts(),
);
assert!(total.count > 0);
assert!(
p99.is_some_and(|ms| ms <= P99_BOUND_MS),
"p99 propagation above {P99_BOUND_MS} ms"
);
}
/// The upper bound of the bucket in which the `quantile` of changes is reached; None if
/// only in the last (unbounded) one.
fn percentile_bound(propagation: &Propagation, quantile: f64) -> Option<u64> {
let wanted = (propagation.count as f64 * quantile).ceil() as u64;
let mut seen = 0;
for (bound, count) in propagation.buckets() {
seen += count;
if seen >= wanted {
return bound;
}
}
None
}
|
