fix: send Event::Connected only after transition to Approved

src/instance/stage/joining.rs

@@ -9,7 +9,7 @@
 use super::{approved::Approved, Command, State};
 use crate::{
     error::{Error, Result},
-    event::{Connected, Event},
+    event::Connected,
     messages::{BootstrapResponse, JoinRequest, Message, Variant, VerifyStatus},
     node::Node,
     peer::Peer,
@@ -89,7 +89,6 @@ impl Joining {
                 verify_message(&msg, trusted_key)?;
 
                 // Transition from Joining to Approved
-                let connect_type = self.connect_type();
                 let section_chain = msg.proof_chain()?.clone();
 
                 info!(
@@ -101,8 +100,6 @@ impl Joining {
                 let state = Approved::new(section, None, self.node.clone());
                 let state = State::Approved(state);
 
-                self.node.send_event(Event::Connected(connect_type));
-
                 Ok(vec![Command::Transition(Box::new(state))])
             }
             _ => {
@@ -146,7 +143,7 @@ impl Joining {
     }
 
     // Are we relocating or joining for the first time?
-    fn connect_type(&self) -> Connected {
+    pub fn connect_type(&self) -> Connected {
         match &self.join_type {
             JoinType::First { .. } => Connected::First,
             JoinType::Relocate(payload) => Connected::Relocate {

src/instance/stage/mod.rs

@@ -297,7 +297,7 @@ impl Stage {
                     Ok(vec![self.handle_schedule_timeout(duration, token).await])
                 }
                 Command::Transition(state) => {
-                    *self.state.lock().await = *state;
+                    self.handle_transition(*state).await;
                     Ok(vec![])
                 }
             }
@@ -394,6 +394,18 @@ impl Stage {
         Command::HandleTimeout(token)
     }
 
+    async fn handle_transition(&self, new_state: State) {
+        let mut old_state = self.state.lock().await;
+
+        // Sending `Event::Connected` here to make sure the transition is complete by the time the
+        // event is received.
+        if let (State::Joining(old), State::Approved(_)) = (&*old_state, &new_state) {
+            old.node.send_event(Event::Connected(old.connect_type()));
+        }
+
+        *old_state = new_state
+    }
+
     async fn log_ident(&self) -> String {
         match &*self.state.lock().await {
             State::Bootstrapping(state) => format!("{}(?) ", state.node.name()),

src/node.rs

@@ -22,6 +22,7 @@ pub(crate) struct Node {
     pub addr: SocketAddr,
     pub age: u8,
     pub network_params: NetworkParams,
+    // TODO: move this event sender somewhere else. This is not an approprate place for it.
     event_tx: mpsc::UnboundedSender<Event>,
 }