src/raft: fix bugs & add logs messages

src/raft/mod.rs

@@ -13,7 +13,7 @@ use datastore::DataStore;
 use p2p::ProtocolRaft;
 pub use raft::Raft;
 
-#[derive(PartialEq, Eq)]
+#[derive(PartialEq, Eq, Debug)]
 pub enum Role {
     Follower,
     Candidate,
@@ -107,9 +107,9 @@ impl Logs {
     }
 }
 
-#[derive(SerialDecodable, SerialEncodable, Clone, Debug, PartialEq, Eq)]
+#[derive(SerialDecodable, SerialEncodable, Clone, Debug)]
 pub struct NetMsg {
-    id: u64,
+    id: u32,
     recipient_id: Option<NodeId>,
     method: NetMsgMethod,
     payload: Vec<u8>,

src/raft/p2p.rs

@@ -6,7 +6,7 @@ use log::debug;
 
 use crate::{net, Result};
 
-use super::{NetMsg, NodeId};
+use super::{NetMsg, NetMsgMethod, NodeId};
 
 pub struct ProtocolRaft {
     id: Option<NodeId>,
@@ -14,7 +14,7 @@ pub struct ProtocolRaft {
     notify_queue_sender: async_channel::Sender<NetMsg>,
     msg_sub: net::MessageSubscription<NetMsg>,
     p2p: net::P2pPtr,
-    msgs: Arc<Mutex<Vec<NetMsg>>>,
+    msgs: Arc<Mutex<Vec<u32>>>,
 }
 
 impl ProtocolRaft {
@@ -46,19 +46,42 @@ impl ProtocolRaft {
 
             debug!(
                 target: "raft",
-                "ProtocolRaft::handle_receive_msg() received {:?}",
-                msg
+                "ProtocolRaft::handle_receive_msg() received id: {:?} method {:?}",
+                &msg.id, &msg.method
             );
 
-            if self.msgs.lock().await.contains(&msg) {
+            if self.msgs.lock().await.contains(&msg.id) {
                 continue
             }
 
+            self.msgs.lock().await.push(msg.id);
+
             let msg = (*msg).clone();
             self.p2p.broadcast(msg.clone()).await?;
 
-            if msg.recipient_id.is_some() && self.id.is_some() && msg.recipient_id != self.id {
-                continue
+            match (self.id.clone(), msg.recipient_id.clone()) {
+                // if the local node and the msg recipient have ids
+                // then check if the ids are equal
+                (Some(id), Some(m_id)) => {
+                    if id != m_id {
+                        continue
+                    }
+                }
+                // if the msg doesn't have a recipient id then the msg is a VoteRequest
+                // and if the local node's id is not None then it can receive
+                // and response with a VoteResponse
+                (Some(_), None) => {}
+                // if the local node's id is None but the recipient's id is not None
+                // then the local node will only handle the msg if its method
+                // is LogRequest
+                (None, Some(_)) => {
+                    if msg.method != NetMsgMethod::LogRequest {
+                        continue
+                    }
+                }
+                // if the local node's id and msg recipient's id are both None then reject
+                // the msg becuase the local node doesn't have the right to vote
+                (None, None) => continue,
             }
 
             self.notify_queue_sender.send(msg).await?;

src/raft/raft.rs

@@ -6,8 +6,8 @@ use std::{cmp::min, collections::HashMap, net::SocketAddr, path::PathBuf, time::
 
 use async_executor::Executor;
 use futures::{select, FutureExt};
-use log::error;
-use rand::Rng;
+use log::{error, info};
+use rand::{rngs::OsRng, Rng, RngCore};
 
 use crate::{
     net,
@@ -125,7 +125,7 @@ impl<T: Decodable + Encodable + Clone> Raft<T> {
 
         let self_id = self.id.clone();
         registry
-            .register(!net::SESSION_SEED, move |channel, p2p| {
+            .register(net::SESSION_ALL, move |channel, p2p| {
                 let self_id = self_id.clone();
                 let sender = p2p_snd.clone();
                 async move { ProtocolRaft::init(self_id, channel, sender, p2p).await }
@@ -135,25 +135,29 @@ impl<T: Decodable + Encodable + Clone> Raft<T> {
         // P2p performs seed session
         p2p.clone().start(executor.clone()).await?;
 
-        executor.spawn(p2p.clone().run(executor.clone())).detach();
+        let executor_cloned = executor.clone();
+        executor_cloned.spawn(p2p.clone().run(executor.clone())).detach();
 
         let p2p_cloned = p2p.clone();
         let p2p_recv = self.sender.1.clone();
-        executor
-            .spawn(async move {
-                loop {
-                    let msg: NetMsg = p2p_recv.recv().await.unwrap();
-                    p2p_cloned.broadcast(msg).await.unwrap();
+        executor.spawn(async move {
+            loop {
+                let msg: NetMsg = p2p_recv.recv().await.unwrap();
+                match p2p_cloned.broadcast(msg).await {
+                    Ok(_) => {}
+                    Err(e) => error!(target: "raft", "error occurred during broadcasting a msg: {}", e) 
                 }
-            })
-            .detach();
+            }
+        }).detach();
 
         let self_nodes = self.nodes.clone();
+        let p2p_cloned = p2p.clone();
         executor
             .spawn(async move {
                 loop {
-                    task::sleep(Duration::from_millis(TIMEOUT_NODES)).await;
-                    let hosts = p2p.hosts().clone();
+                    info!(target: "raft", "load node ids from p2p hosts ips");
+                    task::sleep(Duration::from_millis(TIMEOUT_NODES * 10)).await;
+                    let hosts = p2p_cloned.hosts().clone();
                     let nodes_ip = hosts.load_all().await.clone();
                     let mut nodes = self_nodes.lock().await;
                     for ip in nodes_ip.iter() {
@@ -216,6 +220,8 @@ impl<T: Decodable + Encodable + Clone> Raft<T> {
             )
             .await?;
         }
+
+        info!(target: "raft", "{} {:?}  broadcast a msg", self.id.is_some(), self.role);
         Ok(())
     }
 
@@ -243,6 +249,12 @@ impl<T: Decodable + Encodable + Clone> Raft<T> {
                 self.broadcast_msg(&d).await?;
             }
         }
+
+        info!(
+            target: "raft",
+            "{} {:?}  receive msg id: {}  recipient_id: {:?} method: {:?} ",
+            self.id.is_some(), self.role, msg.id, &msg.recipient_id.is_some(), &msg.method
+        );
         Ok(())
     }
     async fn send(
@@ -251,9 +263,17 @@ impl<T: Decodable + Encodable + Clone> Raft<T> {
         payload: &[u8],
         method: NetMsgMethod,
     ) -> Result<()> {
-        let rnd = rand::random();
-        let net_msg = NetMsg { id: rnd, recipient_id, payload: payload.to_vec(), method };
+        let random_id = OsRng.next_u32();
+
+        info!(
+            target: "raft",
+            "{} {:?}  send a msg id: {}  recipient_id: {:?} method: {:?} ",
+            self.id.is_some(), self.role, random_id, &recipient_id.is_some(), &method
+        );
+
+        let net_msg = NetMsg { id: random_id, recipient_id, payload: payload.to_vec(), method };
         self.sender.0.send(net_msg).await?;
+
         Ok(())
     }
 
@@ -268,7 +288,7 @@ impl<T: Decodable + Encodable + Clone> Raft<T> {
     }
 
     async fn send_vote_request(&mut self) -> Result<()> {
-        // this will prevent the node to become a candidate
+        // this will prevent the listener node to become a candidate
         if self.id.is_none() {
             return Ok(())
         }