diff options
| -rw-r--r-- | src/simtest/timing.rs | 35 | ||||
| -rw-r--r-- | src/transport/sim.rs | 8 |
2 files changed, 40 insertions, 3 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)] diff --git a/src/transport/sim.rs b/src/transport/sim.rs index 3e0163f..1556d69 100644 --- a/src/transport/sim.rs +++ b/src/transport/sim.rs @@ -25,7 +25,7 @@ use tokio::sync::{mpsc, oneshot}; use tracing::{debug, warn}; use super::mycelium::{decode_message, encode_message}; -use super::{EncodedMessage, MessageHandler, ParsedMessage, Transport}; +use super::{EncodedMessage, MessageHandler, MessageKind, ParsedMessage, Transport}; /// How one direction of a link behaves. #[derive(Clone, Debug)] @@ -36,6 +36,9 @@ pub(crate) struct LinkParams { pub(crate) duplicate: f64, /// Delay, uniform in this range. Overlapping delays reorder messages. pub(crate) delay: (Duration, Duration), + /// Messages of these kinds are always lost: a peer that answers some messages but + /// not others (e.g. keepalives but not sync). + pub(crate) drop: Vec<MessageKind>, } impl Default for LinkParams { @@ -45,6 +48,7 @@ impl Default for LinkParams { loss: 0.0, duplicate: 0.0, delay: (Duration::from_millis(1), Duration::from_millis(5)), + drop: Vec::new(), } } } @@ -155,7 +159,7 @@ impl SimNet { .get(&(from.to_owned(), message.peer.clone())) .unwrap_or(&inner.default_link) .clone(); - if inner.rng.chance(params.loss) { + if params.drop.contains(&message.kind) || inner.rng.chance(params.loss) { inner.counts.lost += 1; return; } |