add remake tests on the simulator

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)));
+}