summaryrefslogtreecommitdiffstats
path: root/src
diff options
context:
space:
mode:
authortv <tv@krebsco.de>2026-10-26 21:36:09 +0100
committertv <tv@krebsco.de>2026-10-26 21:36:09 +0100
commit137b1547f847ff2402ea9e634b64ee2196a75075 (patch)
treec6bce410e1c8cb2434be268275220df2afa18fdb /src
parentf1f367d1e2438c87e1e90df14d56aa10f2bd7b8b (diff)
add remake tests on the simulator
Diffstat (limited to 'src')
-rw-r--r--src/remake/api.rs55
-rw-r--r--src/remake/mod.rs2
-rw-r--r--src/simtest/mod.rs1
-rw-r--r--src/simtest/remakes.rs229
4 files changed, 287 insertions, 0 deletions
diff --git a/src/remake/api.rs b/src/remake/api.rs
index 0c2ab10..309bdbd 100644
--- a/src/remake/api.rs
+++ b/src/remake/api.rs
@@ -1009,3 +1009,58 @@ pub(crate) async fn epochs<H: Host>(State(host): State<H>) -> Response {
}))
.into_response()
}
+
+/// For tests on the simulator (`simtest`): the remake endpoints without HTTP, answering
+/// with the status and the JSON body.
+#[cfg(test)]
+pub(crate) mod testing {
+ use super::*;
+
+ async fn into_json(response: Response) -> (StatusCode, JsonValue) {
+ let status = response.status();
+ let bytes = axum::body::to_bytes(response.into_body(), usize::MAX)
+ .await
+ .expect("response body");
+ (
+ status,
+ serde_json::from_slice(&bytes).unwrap_or(JsonValue::Null),
+ )
+ }
+
+ /// `POST /admin/remakes` with `body` and `?force=`.
+ pub(crate) async fn start<H: Host>(
+ host: H,
+ body: JsonValue,
+ force: Option<&str>,
+ ) -> (StatusCode, JsonValue) {
+ let query = StartQuery {
+ force: force.map(str::to_owned),
+ ..StartQuery::default()
+ };
+ let mut headers = HeaderMap::new();
+ headers.insert(
+ CONTENT_TYPE,
+ axum::http::HeaderValue::from_static("application/json"),
+ );
+ into_json(start_kind(host, query, &headers, Bytes::from(body.to_string())).await).await
+ }
+
+ /// `DELETE /admin/remakes/{id}` with `?force=`.
+ pub(crate) async fn abort<H: Host>(
+ host: H,
+ id: &str,
+ force: Option<&str>,
+ ) -> (StatusCode, JsonValue) {
+ let query = AbortQuery {
+ wait: false,
+ force: force.map(str::to_owned),
+ };
+ into_json(abort_by_id(host, id.to_owned(), query).await).await
+ }
+
+ /// `GET /admin/remakes/{id}`: the report, if the remake exists here.
+ pub(crate) async fn report<H: Host>(host: &H, id: &str) -> Option<JsonValue> {
+ let report = load_report(host, id).await?;
+ Some(serde_json::to_value(report).expect("report"))
+ }
+}
diff --git a/src/remake/mod.rs b/src/remake/mod.rs
index bd04de9..79e0a1b 100644
--- a/src/remake/mod.rs
+++ b/src/remake/mod.rs
@@ -32,6 +32,8 @@ pub(crate) mod snapshot;
mod switch;
mod vote;
+#[cfg(test)]
+pub(crate) use api::testing;
pub(crate) use kind::remake_genesis;
pub(crate) use local::RemakeState;
pub(crate) use switch::run_worker;
diff --git a/src/simtest/mod.rs b/src/simtest/mod.rs
index 3202ae5..f7af61b 100644
--- a/src/simtest/mod.rs
+++ b/src/simtest/mod.rs
@@ -9,6 +9,7 @@
//! `RUST_LOG=continuum=debug cargo test simtest -- --nocapture` shows the nodes' logs
//! (all nodes interleaved; messages name peers by ID).
+mod remakes;
mod soak;
mod timing;
diff --git a/src/simtest/remakes.rs b/src/simtest/remakes.rs
new file mode 100644
index 0000000..28f3768
--- /dev/null
+++ b/src/simtest/remakes.rs
@@ -0,0 +1,229 @@
+//! Remakes on whole nodes (`docs/remakes.md`): a compaction everywhere, competing
+//! proposals across a partition, a node that misses the switch, a coordinator that is
+//! gone, and a skewed clock.
+//!
+//! The simulated nodes' timings make a remake quick: settle 2 s (two `dormant_after`),
+//! switch delay a keepalive, votes within half a keepalive.
+
+use std::time::Duration;
+
+use axum::http::StatusCode;
+use serde_json::{Value as JsonValue, json};
+
+use super::{Sim, read, write};
+use crate::document::replica::Epoch;
+use crate::remake::testing;
+use crate::state::AppState;
+use crate::swarm::bootstrap::Start;
+use crate::transport::sim::LinkParams;
+
+/// Joining, registering and converging.
+const SOON: Duration = Duration::from_secs(10);
+/// A remake from proposal to every node switched, rejoins included.
+const REMAKE: Duration = Duration::from_secs(30);
+
+/// Three nodes, ready, registered in the member list (so in the electorate) and in step.
+async fn three(sim: &Sim) -> (AppState, AppState, AppState) {
+ let a = sim.start("a", &[], Start::Genesis);
+ let b = sim.start("b", &["a"], Start::Join);
+ let c = sim.start("c", &["a"], Start::Join);
+ sim.wait_ready(&[&a, &b, &c], SOON).await;
+ let nodes = [a.clone(), b.clone(), c.clone()];
+ sim.wait_until("every node is registered", SOON, || {
+ let nodes = nodes.clone();
+ async move {
+ for node in &nodes {
+ let members = crate::swarm::members::members(&node.doc.read().await.doc);
+ if members.len() < 3 {
+ return false;
+ }
+ }
+ true
+ }
+ })
+ .await;
+ (a, b, c)
+}
+
+async fn propose(node: &AppState, body: JsonValue) -> String {
+ let (status, report) = testing::start(node.clone(), body, None).await;
+ assert!(status.is_success(), "proposal refused: {status} {report}");
+ report["id"].as_str().expect("remake id").to_owned()
+}
+
+async fn on_generation(sim: &Sim, nodes: &[&AppState], generation: u64) {
+ sim.wait_until(
+ &format!("every node is on generation {generation}"),
+ REMAKE,
+ move || async move {
+ for node in nodes {
+ if node.local_epoch().generation != generation || !node.is_ready().await {
+ return false;
+ }
+ }
+ true
+ },
+ )
+ .await;
+}
+
+fn same_epoch(nodes: &[&AppState]) -> Epoch {
+ let epoch = nodes[0].local_epoch();
+ for node in nodes {
+ assert_eq!(node.local_epoch(), epoch);
+ }
+ epoch
+}
+
+/// A compaction moves every node to the next epoch with the data as it was, and the
+/// swarm works on in it.
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn compaction_switches_every_node() {
+ let sim = Sim::new(31);
+ let (a, b, c) = three(&sim).await;
+ write(&b, "before", json!(1)).await;
+ sim.wait_converged(&[&a, &b, &c], SOON).await;
+
+ propose(&a, json!({ "kind": "compact" })).await;
+ on_generation(&sim, &[&a, &b, &c], 1).await;
+ same_epoch(&[&a, &b, &c]);
+ for node in [&a, &b, &c] {
+ assert_eq!(read(node, "before").await, Some(json!(1)));
+ }
+
+ write(&c, "after", json!(2)).await;
+ let a = &a;
+ sim.wait_until(
+ "a has what c wrote in the new epoch",
+ SOON,
+ move || async move { read(a, "after").await == Some(json!(2)) },
+ )
+ .await;
+}
+
+/// Two proposals on two sides of a partition: the side with the majority commits, the
+/// other's can't, and once the partition heals every node is on the one new epoch.
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn competing_proposals_commit_at_most_once() {
+ let sim = Sim::new(32);
+ let (a, b, c) = three(&sim).await;
+ write(&a, "data", json!("x")).await;
+ sim.wait_converged(&[&a, &b, &c], SOON).await;
+
+ let cut = LinkParams {
+ loss: 1.0,
+ ..LinkParams::default()
+ };
+ for other in ["b", "c"] {
+ sim.net.set_link("a", other, cut.clone());
+ sim.net.set_link(other, "a", cut.clone());
+ }
+ let lonely = propose(&a, json!({ "kind": "compact" })).await;
+ propose(&b, json!({ "kind": "compact" })).await;
+ on_generation(&sim, &[&b, &c], 1).await;
+ assert_eq!(a.local_epoch().generation, 0, "a's remake had no majority");
+
+ for other in ["b", "c"] {
+ sim.net.set_link("a", other, LinkParams::default());
+ sim.net.set_link(other, "a", LinkParams::default());
+ }
+ on_generation(&sim, &[&a, &b, &c], 1).await;
+ same_epoch(&[&a, &b, &c]);
+ assert_eq!(read(&a, "data").await, Some(json!("x")));
+ // a's own remake never committed (its record stayed in the old epoch).
+ assert!(
+ testing::report(&a, &lonely)
+ .await
+ .is_none_or(|report| report["phase"] != "committed")
+ );
+}
+
+/// A node that is away during the remake catches up with the new epoch when it
+/// returns, and what it wrote meanwhile isn't lost.
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn node_away_during_the_switch_rejoins() {
+ let sim = Sim::new(33);
+ let (a, b, c) = three(&sim).await;
+ write(&a, "data", json!(1)).await;
+ sim.wait_converged(&[&a, &b, &c], SOON).await;
+
+ sim.net.take_down("c");
+ write(&c, "while/away", json!(true)).await;
+ // c counts in the electorate (its alive stamp is recent); a and b are a majority.
+ propose(&a, json!({ "kind": "compact" })).await;
+ on_generation(&sim, &[&a, &b], 1).await;
+ assert_eq!(c.local_epoch().generation, 0);
+
+ sim.net.bring_up("c");
+ on_generation(&sim, &[&a, &b, &c], 1).await;
+ same_epoch(&[&a, &b, &c]);
+ assert_eq!(read(&c, "data").await, Some(json!(1)));
+ let a = &a;
+ sim.wait_until(
+ "a has what c wrote while away",
+ REMAKE,
+ move || async move { read(a, "while/away").await == Some(json!(true)) },
+ )
+ .await;
+}
+
+/// A coordinator that is gone before it decides: its remake expires, another node
+/// aborts it (which frees the votes), and a new proposal commits without it.
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn remake_of_a_gone_coordinator_expires_and_is_aborted() {
+ let sim = Sim::new(34);
+ let (a, b, c) = three(&sim).await;
+
+ // The proposal reaches b and c, but their votes never reach a.
+ let cut = LinkParams {
+ loss: 1.0,
+ ..LinkParams::default()
+ };
+ sim.net.set_link("b", "a", cut.clone());
+ sim.net.set_link("c", "a", cut);
+ let orphan = propose(&a, json!({ "kind": "compact" })).await;
+ let (b_ref, c_ref, orphan_ref) = (&b, &c, orphan.as_str());
+ sim.wait_until("b has the proposal", SOON, move || async move {
+ testing::report(b_ref, orphan_ref).await.is_some()
+ })
+ .await;
+ sim.net.take_down("a");
+
+ // Expired after its settle period plus a grace period, abortable a grace period
+ // later (until then the answer is 409).
+ sim.wait_until(
+ "the orphaned remake can be aborted",
+ REMAKE,
+ move || async move {
+ let (status, _) = testing::abort(b_ref.clone(), orphan_ref, None).await;
+ status == StatusCode::OK
+ },
+ )
+ .await;
+ sim.wait_until("c sees it aborted", SOON, move || async move {
+ testing::report(c_ref, orphan_ref)
+ .await
+ .is_some_and(|report| report["phase"] == "aborted")
+ })
+ .await;
+
+ propose(&b, json!({ "kind": "compact" })).await;
+ on_generation(&sim, &[&b, &c], 1).await;
+ same_epoch(&[&b, &c]);
+}
+
+/// A node whose clock is ahead (within the grace period) takes part as any other.
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn compaction_with_a_skewed_clock() {
+ let sim = Sim::new(35);
+ let (a, b, c) = three(&sim).await;
+ // dormant_after (the grace period) is 1 s here.
+ c.clock.set_offset_ms(700);
+ write(&a, "data", json!(1)).await;
+ sim.wait_converged(&[&a, &b, &c], SOON).await;
+
+ propose(&c, json!({ "kind": "compact" })).await;
+ on_generation(&sim, &[&a, &b, &c], 1).await;
+ same_epoch(&[&a, &b, &c]);
+ assert_eq!(read(&b, "data").await, Some(json!(1)));
+}