aboutsummaryrefslogtreecommitdiffstats
path: root/src/simtest/remakes.rs
diff options
context:
space:
mode:
Diffstat (limited to 'src/simtest/remakes.rs')
-rw-r--r--src/simtest/remakes.rs355
1 files changed, 351 insertions, 4 deletions
diff --git a/src/simtest/remakes.rs b/src/simtest/remakes.rs
index 28f3768..6a1f0a9 100644
--- a/src/simtest/remakes.rs
+++ b/src/simtest/remakes.rs
@@ -1,6 +1,10 @@
//! 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.
+//! gone, a skewed clock, writes while a node starts to rejoin, and two remakes committed
+//! for one epoch (a coordinator cut off after it committed; electorates of a partition
+//! older than the electorate window; a vote lost with a restart), and a node with its
+//! clock ahead that would abort
+//! a remake as expired while its coordinator still runs it.
//!
//! The simulated nodes' timings make a remake quick: settle 2 s (two `dormant_after`),
//! switch delay a keepalive, votes within half a keepalive.
@@ -10,15 +14,15 @@ use std::time::Duration;
use axum::http::StatusCode;
use serde_json::{Value as JsonValue, json};
-use super::{Sim, read, write};
+use super::{SOON, Sim, read, write};
+use crate::document::ChangeHash;
use crate::document::replica::Epoch;
+use crate::document::store;
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);
@@ -227,3 +231,346 @@ async fn compaction_with_a_skewed_clock() {
same_epoch(&[&a, &b, &c]);
assert_eq!(read(&b, "data").await, Some(json!(1)));
}
+
+/// Writes on a node through its rejoin: every one is answered and kept, those during
+/// the rejoin in the document before it (`StaleView`) and carried over at the end;
+/// none lands in the fresh document before the epoch's snapshot, which every change of
+/// the new epoch descends from.
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn writes_through_a_rejoin_are_kept() {
+ use std::sync::atomic::{AtomicBool, Ordering};
+ use std::sync::{Arc, Mutex};
+
+ let sim = Sim::new(36);
+ let (a, b, c) = three(&sim).await;
+ sim.net.take_down("c");
+ propose(&a, json!({ "kind": "compact" })).await;
+ on_generation(&sim, &[&a, &b], 1).await;
+
+ // c writes on and on, through its return and rejoin.
+ let answered = Arc::new(Mutex::new(Vec::<u64>::new()));
+ let stop = Arc::new(AtomicBool::new(false));
+ let writer = {
+ let (c, answered, stop) = (c.clone(), Arc::clone(&answered), Arc::clone(&stop));
+ tokio::spawn(async move {
+ let mut n = 0_u64;
+ while !stop.load(Ordering::Relaxed) {
+ let at = vec!["w".to_owned(), n.to_string()];
+ let written = c
+ .mutate_doc(move |doc| store::write_path(doc, &at, &json!(n)))
+ .await;
+ assert!(written.is_ok(), "write {n} refused: {written:?}");
+ answered.lock().unwrap().push(n);
+ n += 1;
+ tokio::time::sleep(Duration::from_millis(2)).await;
+ }
+ })
+ };
+ tokio::time::sleep(Duration::from_millis(200)).await;
+ sim.net.bring_up("c");
+ on_generation(&sim, &[&a, &b, &c], 1).await;
+ tokio::time::sleep(Duration::from_millis(200)).await;
+ stop.store(true, Ordering::Relaxed);
+ writer.await.unwrap();
+
+ // Every change of c's document descends from the epoch's snapshot.
+ let epoch = same_epoch(&[&a, &b, &c]);
+ {
+ let replica = c.doc.read().await;
+ let snapshot = ChangeHash(epoch.id);
+ let deps: std::collections::HashMap<ChangeHash, Vec<ChangeHash>> = replica
+ .doc
+ .get_changes(&[])
+ .iter()
+ .map(|change| (change.hash(), change.deps().to_vec()))
+ .collect();
+ for start in deps.keys() {
+ let (mut todo, mut seen) = (vec![*start], std::collections::HashSet::new());
+ let mut reached = false;
+ while let Some(hash) = todo.pop() {
+ if hash == snapshot {
+ reached = true;
+ break;
+ }
+ if seen.insert(hash) {
+ todo.extend(deps.get(&hash).into_iter().flatten().copied());
+ }
+ }
+ assert!(reached, "change {start} doesn't descend from the snapshot");
+ }
+ }
+ // Every write is there (those before and during the rejoin replayed into it).
+ let answered = answered.lock().unwrap().clone();
+ assert!(!answered.is_empty());
+ let c = &c;
+ sim.wait_until("c has every write it answered", SOON, move || {
+ let answered = answered.clone();
+ async move {
+ for n in &answered {
+ if read(c, &format!("w/{n}")).await != Some(json!(n)) {
+ return false;
+ }
+ }
+ true
+ }
+ })
+ .await;
+}
+
+/// A coordinator cut off right after it committed: the others never hear of the commit,
+/// abort the remake once it expired and commit another. The coordinator switches to its
+/// own; once back, it sees more peers on the other epoch of that generation, rejoins
+/// that one sideways (`catchup::sideways`), and the swarm is on one epoch again.
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn coordinator_cut_off_after_committing_rejoins_the_others() {
+ let sim = Sim::new(37);
+ let (a, b, c) = three(&sim).await;
+ write(&b, "data", json!(1)).await;
+ sim.wait_converged(&[&a, &b, &c], SOON).await;
+
+ let cut = LinkParams {
+ loss: 1.0,
+ ..LinkParams::default()
+ };
+ let lost = propose(&a, json!({ "kind": "compact" })).await;
+ let (a_ref, b_ref, c_ref, lost_ref) = (&a, &b, &c, lost.as_str());
+ sim.wait_until("b and c have the proposal", SOON, move || async move {
+ testing::report(b_ref, lost_ref).await.is_some()
+ && testing::report(c_ref, lost_ref).await.is_some()
+ })
+ .await;
+ // Their votes still reach a; nothing from a reaches them.
+ for other in ["b", "c"] {
+ sim.net.set_link("a", other, cut.clone());
+ }
+ sim.wait_until("a committed", SOON, move || async move {
+ testing::report(a_ref, lost_ref)
+ .await
+ .is_some_and(|report| report["phase"] == "committed")
+ })
+ .await;
+ sim.net.take_down("a");
+ on_generation(&sim, &[&a], 1).await;
+
+ // For b and c, the remake expired undecided.
+ sim.wait_until("b can abort the lost remake", REMAKE, move || async move {
+ let (status, _) = testing::abort(b_ref.clone(), lost_ref, None).await;
+ status == StatusCode::OK
+ })
+ .await;
+ propose(&b, json!({ "kind": "compact" })).await;
+ on_generation(&sim, &[&b, &c], 1).await;
+ let theirs = same_epoch(&[&b, &c]);
+ assert_ne!(a.local_epoch(), theirs, "two epochs of generation 1");
+
+ for other in ["b", "c"] {
+ sim.net.set_link("a", other, LinkParams::default());
+ }
+ sim.net.bring_up("a");
+ sim.wait_until("a is on b's and c's epoch", REMAKE, move || async move {
+ a_ref.local_epoch() == theirs && a_ref.is_ready().await
+ })
+ .await;
+ assert_eq!(read(&a, "data").await, Some(json!(1)));
+}
+
+/// Electorates that don't overlap: on one side of a partition the other side's alive
+/// stamps look older than `--electorate-window` (here by the clocks of d and e, a month
+/// ahead, as a partition that long would make them), so each side's remake gets a
+/// majority of its own electorate and both commit. Once the partition heals, the two
+/// nodes of the smaller side rejoin the larger side's epoch.
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn both_sides_of_an_old_partition_commit_and_converge() {
+ use crate::remake::Host;
+
+ let sim = Sim::new(38);
+ let a = sim.start("a", &[], Start::Genesis);
+ let [b, c, d, e] = ["b", "c", "d", "e"].map(|id| sim.start(id, &["a"], Start::Join));
+ let all = [&a, &b, &c, &d, &e];
+ sim.wait_ready(&all, SOON).await;
+ let nodes = [a.clone(), b.clone(), c.clone(), d.clone(), e.clone()];
+ sim.wait_until("every node is registered", SOON, || {
+ let nodes = nodes.clone();
+ async move {
+ for node in &nodes {
+ if crate::swarm::members::members(&node.doc.read().await.doc).len() < 5 {
+ return false;
+ }
+ }
+ true
+ }
+ })
+ .await;
+ write(&a, "data", json!(1)).await;
+ sim.wait_converged(&all, SOON).await;
+
+ let cut = LinkParams {
+ loss: 1.0,
+ ..LinkParams::default()
+ };
+ let link = |params: &LinkParams| {
+ for one in ["a", "b", "c"] {
+ for other in ["d", "e"] {
+ sim.net.set_link(one, other, params.clone());
+ sim.net.set_link(other, one, params.clone());
+ }
+ }
+ };
+ link(&cut);
+ let month = 31 * 24 * 3600 * 1000;
+ d.clock.set_offset_ms(month);
+ e.clock.set_offset_ms(month);
+ // d and e renew their stamps (by their clocks, theirs were a month old), and count
+ // only themselves.
+ let d_ref = &d;
+ sim.wait_until("d's electorate is d and e", SOON, move || async move {
+ let doc = d_ref.doc.read().await;
+ d_ref.electorate(&doc.doc, d_ref.now_millis()).len() == 2
+ })
+ .await;
+
+ propose(&a, json!({ "kind": "compact" })).await;
+ propose(&d, json!({ "kind": "compact" })).await;
+ on_generation(&sim, &[&a, &b, &c], 1).await;
+ on_generation(&sim, &[&d, &e], 1).await;
+ let larger = same_epoch(&[&a, &b, &c]);
+ assert_ne!(same_epoch(&[&d, &e]), larger, "two epochs of generation 1");
+
+ link(&LinkParams::default());
+ let (d_ref, e_ref) = (&d, &e);
+ sim.wait_until("d and e are on a's epoch", REMAKE, move || async move {
+ d_ref.local_epoch() == larger
+ && e_ref.local_epoch() == larger
+ && d_ref.is_ready().await
+ && e_ref.is_ready().await
+ })
+ .await;
+ assert_eq!(read(&e, "data").await, Some(json!(1)));
+}
+
+/// A node whose clock is ahead sees a remake expire early: aborting it from there is
+/// refused while the node knows its clock is off (`?force=clock` overrides), as its
+/// coordinator, alive, may still commit it.
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn abort_from_a_node_with_its_clock_off_is_refused() {
+ use crate::remake::Host;
+
+ // Under the identity layer: its paths layer compares the peers' clocks (PONGs).
+ let sim = Sim::with_identity(39);
+ let (a, b, c) = three(&sim).await;
+ // Far beyond the grace period (1 s) and --clock-skew-warn (2 s), and the settle
+ // period below: by c's clock, the remake is expired once proposed.
+ c.clock.set_offset_ms(60_000);
+ let c_ref = &c;
+ sim.wait_until("c knows its clock is off", SOON, move || async move {
+ c_ref.clock_skew().await.off
+ })
+ .await;
+
+ // No vote reaches a, so its remake stays undecided for its settle period.
+ let cut = LinkParams {
+ loss: 1.0,
+ ..LinkParams::default()
+ };
+ sim.net.set_link("b", "a", cut.clone());
+ sim.net.set_link("c", "a", cut);
+ let id = propose(&a, json!({ "kind": "compact", "settle": "30s" })).await;
+ let id_ref = id.as_str();
+ sim.wait_until(
+ "c has the proposal, expired by its clock",
+ SOON,
+ move || async move {
+ testing::report(c_ref, id_ref)
+ .await
+ .is_some_and(|report| report["phase"] == "expired")
+ },
+ )
+ .await;
+ let (status, refused) = testing::abort(c.clone(), &id, None).await;
+ assert_eq!(status, StatusCode::CONFLICT, "{refused}");
+ assert_eq!(refused["force"], "clock");
+ assert!(
+ testing::report(&b, &id)
+ .await
+ .is_some_and(|report| report["phase"] != "aborted")
+ );
+}
+
+/// A vote lost with a restart: v votes on one side of a partition, restarts without a
+/// document file, joins the other side, which never saw the remake, and votes there
+/// for another. Each side reaches its majority with v, and both commit. Once the
+/// partition heals, the side fewer nodes are on rejoins the other's epoch sideways.
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn a_vote_lost_with_a_restart_and_the_swarm_converges() {
+ let sim = Sim::new(40);
+ let a = sim.start("a", &[], Start::Genesis);
+ let [b, v, d, e] = ["b", "v", "d", "e"].map(|id| sim.start(id, &["a"], Start::Join));
+ sim.wait_ready(&[&a, &b, &v, &d, &e], SOON).await;
+ let nodes = [a.clone(), b.clone(), v.clone(), d.clone(), e.clone()];
+ sim.wait_until("every node is registered", SOON, || {
+ let nodes = nodes.clone();
+ async move {
+ for node in &nodes {
+ if crate::swarm::members::members(&node.doc.read().await.doc).len() < 5 {
+ return false;
+ }
+ }
+ true
+ }
+ })
+ .await;
+ write(&a, "data", json!(1)).await;
+ sim.wait_converged(&[&a, &b, &v, &d, &e], SOON).await;
+
+ let cut = LinkParams {
+ loss: 1.0,
+ ..LinkParams::default()
+ };
+ let link = |side: &[&str], other: &[&str], params: &LinkParams| {
+ for one in side {
+ for two in other {
+ sim.net.set_link(one, two, params.clone());
+ sim.net.set_link(two, one, params.clone());
+ }
+ }
+ };
+ link(&["a", "b", "v"], &["d", "e"], &cut);
+
+ // a, b and v: three of five, X commits.
+ let x = propose(&a, json!({ "kind": "compact" })).await;
+ let (a_ref, x_ref) = (&a, x.as_str());
+ sim.wait_until("X committed", REMAKE, move || async move {
+ testing::report(a_ref, x_ref)
+ .await
+ .is_some_and(|report| report["phase"] == "committed")
+ })
+ .await;
+
+ // v restarts on the other side, with nothing of what it had: no record of X.
+ link(&["v"], &["a", "b"], &cut);
+ link(&["v"], &["d", "e"], &LinkParams::default());
+ let v = sim.restart_with("v", &["d"], Start::Join, |_| {});
+ sim.wait_ready(&[&v], SOON).await;
+
+ // d, e and v: three of five, Y commits too.
+ propose(&d, json!({ "kind": "compact" })).await;
+ on_generation(&sim, &[&a, &b], 1).await;
+ on_generation(&sim, &[&d, &e, &v], 1).await;
+ let more = same_epoch(&[&d, &e, &v]);
+ assert_ne!(same_epoch(&[&a, &b]), more, "two epochs of generation 1");
+
+ link(&["a", "b"], &["v", "d", "e"], &LinkParams::default());
+ let (b_ref, more_ref) = (&b, more);
+ sim.wait_until(
+ "a and b are on the epoch more nodes are on",
+ REMAKE,
+ move || async move {
+ a_ref.local_epoch() == more_ref
+ && b_ref.local_epoch() == more_ref
+ && a_ref.is_ready().await
+ && b_ref.is_ready().await
+ },
+ )
+ .await;
+ assert_eq!(read(&b, "data").await, Some(json!(1)));
+}