diff options
| author | tv <tv@krebsco.de> | 2026-11-01 19:57:54 +0100 |
|---|---|---|
| committer | tv <tv@krebsco.de> | 2026-11-01 19:57:54 +0100 |
| commit | 7993cdfe8f8e24b39091b10a0a1bd45d66e3ccd8 (patch) | |
| tree | 252af9493cb63f4acdfdbf7fac3b3b6ad871fbcc /src/transport | |
| parent | 541d1444efc4150ad3e644c7c262f51a4a26c0bb (diff) | |
add restart tests
Diffstat (limited to 'src/transport')
| -rw-r--r-- | src/transport/sim.rs | 33 |
1 files changed, 28 insertions, 5 deletions
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(()) } |
