add rejoin writable tests

1 files changed, 88 insertions, 1 deletions

diff --git a/src/simtest/remakes.rs b/src/simtest/remakes.rs
index e4f115b..ab04efe 100644
--- a/src/simtest/remakes.rs
+++ b/src/simtest/remakes.rs
@@ -1,6 +1,6 @@
 //! 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, and writes while a node starts to rejoin.
 //!
 //! The simulated nodes' timings make a remake quick: settle 2 s (two `dormant_after`),
 //! switch delay a keepalive, votes within half a keepalive.
@@ -11,7 +11,9 @@ use axum::http::StatusCode;
 use serde_json::{Value as JsonValue, json};
 
 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;
@@ -225,3 +227,88 @@ 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 right as it starts to rejoin (`Replica::writable`): each is either
+/// answered and kept, or refused; 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_while_a_rejoin_starts_are_kept_or_refused() {
+    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;
+                if written.is_ok() {
+                    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 answered is there (those before 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;
+}