add restart tests

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<String>,
     inboxes: HashMap<String, mpsc::UnboundedSender<Delivery>>,
+    /// 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<String, u64>,
     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<SimTransport> {
         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<Option<mpsc::UnboundedReceiver<Delivery>>>,
 }
@@ -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(())
     }