add stalled source test

1 files changed, 34 insertions, 1 deletions

diff --git a/src/simtest/timing.rs b/src/simtest/timing.rs
index df7849f..7810f28 100644
--- a/src/simtest/timing.rs
+++ b/src/simtest/timing.rs
@@ -10,7 +10,7 @@ use crate::document::replica::Epoch;
 use crate::state::{AppState, Liveness};
 use crate::swarm::bootstrap::Start;
 use crate::transport::sim::LinkParams;
-use crate::transport::{HeadsSummary, HelloRequest, Keepalive, ParsedMessage, RunId};
+use crate::transport::{HeadsSummary, HelloRequest, Keepalive, MessageKind, ParsedMessage, RunId};
 
 /// Long enough for anything that should happen within a few keepalives.
 const SOON: Duration = Duration::from_secs(10);
@@ -85,6 +85,7 @@ async fn converges_over_lossy_links() {
         loss: 0.15,
         duplicate: 0.05,
         delay: (Duration::from_millis(1), Duration::from_millis(30)),
+        ..LinkParams::default()
     });
     let a = sim.start("a", &[], Start::Genesis);
     let b = sim.start("b", &["a"], Start::Join);
@@ -107,6 +108,38 @@ async fn converges_over_lossy_links() {
     assert!(sim.net.counts().lost > 0);
 }
 
+/// R24: a catch-up source that is alive (it answers handshakes and keepalives) but whose
+/// sync messages never arrive is given up after three unanswered sync messages and
+/// `delivery_limit` without one, and another peer takes over.
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn stalled_catch_up_source_is_replaced() {
+    let sim = Sim::new(6);
+    let a = sim.start("a", &[], Start::Genesis);
+    let b = sim.start("b", &["a"], Start::Join);
+    let (a, b) = (&a, &b);
+    sim.wait_ready(&[a, b], SOON).await;
+    write(a, "data", json!("x")).await;
+    sim.wait_converged(&[a, b], SOON).await;
+
+    // c joins through a, so a becomes its source (the first ready peer it hears
+    // from); b it learns of from a.
+    sim.net.set_link(
+        "a",
+        "c",
+        LinkParams {
+            drop: vec![MessageKind::Sync],
+            ..LinkParams::default()
+        },
+    );
+    let c = sim.start("c", &["a"], Start::Join);
+    let c = &c;
+    sim.wait_ready(&[c], LOSSY).await;
+
+    assert_eq!(read(c, "data").await, Some(json!("x")));
+    let sources = c.stats.snapshot().catch_up_sources;
+    assert!(sources >= 2, "{sources} catch-up sources");
+}
+
 /// A peer that hears nothing for `dormant_after` is dormant; once messages flow again
 /// it is active, re-handshakes, and syncs what it missed.
 #[tokio::test(flavor = "multi_thread", worker_threads = 4)]