From 7993cdfe8f8e24b39091b10a0a1bd45d66e3ccd8 Mon Sep 17 00:00:00 2001 From: tv Date: Sun, 1 Nov 2026 19:57:54 +0100 Subject: add restart tests --- src/simtest/mod.rs | 24 ++++++++++++++++++++++++ src/simtest/timing.rs | 24 +++++++++++++++++++++++- src/transport/sim.rs | 33 ++++++++++++++++++++++++++++----- 3 files changed, 75 insertions(+), 6 deletions(-) diff --git a/src/simtest/mod.rs b/src/simtest/mod.rs index 81bddb5..61c81d1 100644 --- a/src/simtest/mod.rs +++ b/src/simtest/mod.rs @@ -144,6 +144,30 @@ impl Sim { } } +impl Sim { + /// Restarts node `id`: its process is gone, as if it crashed (`SimNet::restart`: + /// what it still sends is lost), and a new one starts on the same ID, as + /// [`start_with`](Self::start_with) starts one. Nothing carries over but what + /// `adjust` gives it (its node key, under the identity layer, by its name). + pub(crate) fn restart_with( + &self, + id: &str, + seeds: &[&str], + start: Start, + adjust: impl FnOnce(&mut NodeConfig), + ) -> AppState { + { + let mut nodes = self.nodes.lock().unwrap(); + for (_, old) in nodes.iter().filter(|(name, _)| name == id) { + old.closing.send_replace(true); + } + nodes.retain(|(name, _)| name != id); + } + self.net.restart(id); + self.start_with(id, seeds, start, adjust) + } +} + /// Logs to the test output if `RUST_LOG` is set. fn init_logging() { if std::env::var_os("RUST_LOG").is_some() { diff --git a/src/simtest/timing.rs b/src/simtest/timing.rs index fd76a17..daa588f 100644 --- a/src/simtest/timing.rs +++ b/src/simtest/timing.rs @@ -1,5 +1,5 @@ //! 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; @@ -46,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); diff --git a/src/transport/sim.rs b/src/transport/sim.rs index e783743..2f167dc 100644 --- a/src/transport/sim.rs +++ b/src/transport/sim.rs @@ -83,6 +83,9 @@ struct NetInner { links: HashMap<(String, String), LinkParams>, down: HashSet, inboxes: HashMap>, + /// Per node, its current run (`restart`): what a transport of an earlier run sends + /// is lost, as from a process that is gone. + runs: HashMap, counts: NetCounts, } @@ -109,6 +112,7 @@ impl SimNet { links: HashMap::new(), down: HashSet::new(), inboxes: HashMap::new(), + runs: HashMap::new(), counts: NetCounts::default(), })), } @@ -121,15 +125,31 @@ impl SimNet { /// A transport for a new node `id` on this net. pub(crate) fn transport(&self, id: &str) -> Arc { let (tx, rx) = mpsc::unbounded_channel(); - let previous = self.lock().inboxes.insert(id.to_owned(), tx); - assert!(previous.is_none(), "node {id} is already on the net"); + let run = { + let mut inner = self.lock(); + let previous = inner.inboxes.insert(id.to_owned(), tx); + assert!(previous.is_none(), "node {id} is already on the net"); + *inner.runs.entry(id.to_owned()).or_default() + }; Arc::new(SimTransport { id: id.to_owned(), + run, net: self.clone(), inbox: tokio::sync::Mutex::new(Some(rx)), }) } + /// Node `id`'s process is gone, as if it crashed: its transport neither sends nor + /// receives any more (messages on their way to it are lost), and [`transport`] + /// may give `id` a new one, for its next run. + /// + /// [`transport`]: Self::transport + pub(crate) fn restart(&self, id: &str) { + let mut inner = self.lock(); + inner.inboxes.remove(id); + *inner.runs.entry(id.to_owned()).or_default() += 1; + } + /// How links without their own parameters behave. pub(crate) fn set_default_link(&self, params: LinkParams) { self.lock().default_link = params; @@ -157,12 +177,13 @@ impl SimNet { } /// Hands a message to the net: decides now whether and when it arrives. - fn send(&self, from: &str, message: &EncodedMessage) { + fn send(&self, from: &str, run: u64, message: &EncodedMessage) { let delays = { let mut guard = self.lock(); let inner = &mut *guard; inner.counts.sent += 1; - if inner.down.contains(from) || inner.down.contains(&message.peer) { + let gone = inner.runs.get(from).is_some_and(|current| *current != run); + if gone || inner.down.contains(from) || inner.down.contains(&message.peer) { inner.counts.lost += 1; return; } @@ -257,6 +278,8 @@ impl SimNet { /// are node IDs. pub(crate) struct SimTransport { id: String, + /// The run of its node it belongs to (`SimNet::restart`). + run: u64, net: SimNet, inbox: tokio::sync::Mutex>>, } @@ -315,7 +338,7 @@ impl Transport for SimTransport { /// Always accepted: the net loses messages silently, like Mycelium once it has /// taken them. async fn send_encoded(&self, message: &EncodedMessage) -> Result<()> { - self.net.send(&self.id, message); + self.net.send(&self.id, self.run, message); Ok(()) } -- cgit v1.3.1