aboutsummaryrefslogtreecommitdiffstats
path: root/src/simtest/timing.rs
diff options
context:
space:
mode:
Diffstat (limited to 'src/simtest/timing.rs')
-rw-r--r--src/simtest/timing.rs111
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;
+}