add soak test

1 files changed, 101 insertions, 0 deletions

diff --git a/src/simtest/soak.rs b/src/simtest/soak.rs
new file mode 100644
index 0000000..6906238
--- /dev/null
+++ b/src/simtest/soak.rs
@@ -0,0 +1,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
+}