diff options
Diffstat (limited to 'src/simtest/timing.rs')
| -rw-r--r-- | src/simtest/timing.rs | 111 |
1 files changed, 101 insertions, 10 deletions
diff --git a/src/simtest/timing.rs b/src/simtest/timing.rs index 20ad15c..daa588f 100644 --- a/src/simtest/timing.rs +++ b/src/simtest/timing.rs @@ -1,19 +1,17 @@ //! The timing rules of `peers.md` on whole nodes: joining and converging, dormancy and -//! return, and late messages that must not undo what newer ones said. +//! return, a restart, and late messages that must not undo what newer ones said. use std::time::Duration; use serde_json::json; -use super::{Sim, has_neighbour, heads, peer, read, write}; +use super::{SOON, Sim, has_neighbour, heads, peer, read, write}; use crate::document::replica::Epoch; -use crate::state::{AppState, Liveness}; +use crate::state::AppState; use crate::swarm::bootstrap::Start; use crate::transport::sim::LinkParams; use crate::transport::{HeadsSummary, HelloRequest, MessageKind, ParsedMessage, RunId, Status}; -/// Long enough for anything that should happen within a few keepalives. -const SOON: Duration = Duration::from_secs(10); /// With lost messages: several reply timeouts (1 s, doubling) and heals in a row. const LOSSY: Duration = Duration::from_secs(30); @@ -48,6 +46,28 @@ async fn three_nodes_join_and_converge() { } } +/// A node that restarts without a document file (its process gone, a new one on the +/// same ID) joins again and gets the document; what its earlier run sends on is lost. +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn restarted_node_joins_again() { + let sim = Sim::new(9); + let a = sim.start("a", &[], Start::Genesis); + let b = sim.start("b", &["a"], Start::Join); + sim.wait_ready(&[&a, &b], SOON).await; + write(&a, "before", json!(1)).await; + sim.wait_converged(&[&a, &b], SOON).await; + + let b = sim.restart_with("b", &["a"], Start::Join, |_| {}); + sim.wait_ready(&[&b], SOON).await; + let b = &b; + sim.wait_until("b has the document again", SOON, move || async move { + read(b, "before").await == Some(json!(1)) + }) + .await; + write(b, "after", json!(2)).await; + sim.wait_converged(&[&a, b], SOON).await; +} + #[tokio::test(flavor = "multi_thread", worker_threads = 4)] async fn converges_with_duplicated_and_reordered_messages() { let sim = Sim::new(2); @@ -232,11 +252,7 @@ async fn peer_goes_dormant_and_returns() { } async fn is_dormant(node: &AppState, other: &str) -> bool { - peer(node, other, |p| { - matches!(p.liveness, Liveness::Dormant { .. }) - }) - .await - == Some(true) + peer(node, other, |p| p.liveness.is_dormant()).await == Some(true) } /// R27: a resent HELLO (its reply was lost) or a late copy of one doesn't reset the @@ -318,6 +334,7 @@ async fn late_epoch_does_not_flip_a_peers_epoch() { epoch_seq, ready: true, syncs_with_you: false, + leaving: false, }) }; let a = &a; @@ -414,3 +431,77 @@ async fn forgotten_node_that_is_alive_cancels_its_record() { } sim.wait_converged(&[a, b, c], SOON).await; } + +/// A transport that only connects out: it publishes no address, as a libp2p node +/// without a listen address (`topology::refresh`, `outbound`). +struct OutboundOnly(std::sync::Arc<dyn crate::transport::Transport>); + +#[async_trait::async_trait] +impl crate::transport::Transport for OutboundOnly { + fn local_id(&self) -> &str { + self.0.local_id() + } + fn is_peer_id(&self, id: &str) -> bool { + self.0.is_peer_id(id) + } + fn delivery_limit(&self) -> Duration { + self.0.delivery_limit() + } + fn local_addresses(&self) -> Vec<String> { + Vec::new() + } + async fn resolve_seed(&self, seed: &str) -> anyhow::Result<String> { + self.0.resolve_seed(seed).await + } + fn encode( + &self, + peer: &str, + message: &ParsedMessage, + ) -> anyhow::Result<crate::transport::EncodedMessage> { + self.0.encode(peer, message) + } + async fn send_encoded(&self, message: &crate::transport::EncodedMessage) -> anyhow::Result<()> { + self.0.send_encoded(message).await + } + async fn start_consumer( + &self, + handler: crate::transport::MessageHandler, + ) -> anyhow::Result<()> { + self.0.start_consumer(handler).await + } + async fn wait_for_exit(&self) -> std::io::Result<std::process::ExitStatus> { + self.0.wait_for_exit().await + } +} + +/// A member with no address (it only connects out) is in nobody's graph but its own: +/// those it picked keep it as a neighbour, so their changes reach it too, not only +/// answers to its own. +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn a_member_that_only_connects_out_gets_changes() { + let sim = Sim::new(7); + let a = sim.start("a", &[], Start::Genesis); + let b = sim.start_with("b", &["a"], Start::Join, |config| { + config.transport = std::sync::Arc::new(OutboundOnly(config.transport.clone())); + }); + let c = sim.start("c", &["a"], Start::Join); + let (a, b, c) = (&a, &b, &c); + sim.wait_ready(&[a, b, c], SOON).await; + // Once b is listed (without an address), the member list no longer makes it a's + // or c's neighbour. + sim.wait_until("b is listed", SOON, move || async move { + read(a, "_continuum/members/b").await.is_some() + && read(c, "_continuum/members/b").await.is_some() + }) + .await; + sim.wait_until("a and c keep b", SOON, move || async move { + has_neighbour(a, "b") && has_neighbour(c, "b") + }) + .await; + write(a, "from/a", json!(1)).await; + write(c, "from/c", json!(3)).await; + sim.wait_until("b has a's and c's writes", SOON, move || async move { + read(b, "from/a").await == Some(json!(1)) && read(b, "from/c").await == Some(json!(3)) + }) + .await; +} |
