raft: split the main consensus code to files according to the roles & remove sync process

bin/tau/taud/src/main.rs

@@ -172,11 +172,7 @@ async fn realmain(settings: Args, executor: Arc<Executor<'_>>) -> Result<()> {
     let seen_net_msgs = Arc::new(Mutex::new(FxHashMap::default()));
 
     let datastore_raft = datastore_path.join("tau.db");
-    let raft_settings = RaftSettings {
-        datastore_path: datastore_raft,
-        external_addr: net_settings.external_addr.clone(),
-        ..RaftSettings::default()
-    };
+    let raft_settings = RaftSettings { datastore_path: datastore_raft, ..RaftSettings::default() };
 
     let mut raft = Raft::<EncryptedTask>::new(raft_settings, seen_net_msgs.clone())?;
 
@@ -204,7 +200,7 @@ async fn realmain(settings: Args, executor: Arc<Executor<'_>>) -> Result<()> {
 
     let registry = p2p.protocol_registry();
 
-    let raft_node_id = raft.id.clone();
+    let raft_node_id = raft.get_id();
     registry
         .register(net::SESSION_ALL, move |channel, p2p| {
             let raft_node_id = raft_node_id.clone();

script/research/raft-diag/src/main.rs

@@ -91,10 +91,10 @@ fn check(args: Args) -> Result<()> {
 
 async fn start_broadcasting(n: u32, sender: async_channel::Sender<Message>) -> Result<()> {
     sleep(8).await;
-    info!("Start broadcasting...");
+    info!(target: "raft", "Start broadcasting...");
     for id in 0..n {
         let msg = format!("msg_test_{}", id);
-        info!("Send a message {:?}", msg);
+        info!(target: "raft", "Send a message {:?}", msg);
         let msg = Message { payload: msg };
         sender.send(msg).await?;
     }
@@ -105,7 +105,7 @@ async fn start_broadcasting(n: u32, sender: async_channel::Sender<Message>) -> R
 async fn receive_loop(receiver: async_channel::Receiver<Message>) -> Result<()> {
     loop {
         let msg = receiver.recv().await?;
-        info!("Receive new msg {:?}", msg);
+        info!(target: "raft", "Receive new msg {:?}", msg);
     }
 }
 
@@ -126,11 +126,7 @@ async fn start(args: Args, executor: Arc<Executor<'_>>) -> Result<()> {
 
     let seen_net_msgs = Arc::new(Mutex::new(FxHashMap::default()));
 
-    let raft_settings = RaftSettings {
-        datastore_path: datastore_raft,
-        external_addr: net_settings.external_addr.clone(),
-        ..RaftSettings::default()
-    };
+    let raft_settings = RaftSettings { datastore_path: datastore_raft, ..RaftSettings::default() };
 
     let mut raft = Raft::<Message>::new(raft_settings, seen_net_msgs.clone())?;
 
@@ -145,7 +141,7 @@ async fn start(args: Args, executor: Arc<Executor<'_>>) -> Result<()> {
 
     let registry = p2p.protocol_registry();
 
-    let raft_node_id = raft.id.clone();
+    let raft_node_id = raft.get_id();
     registry
         .register(net::SESSION_ALL, move |channel, p2p| {
             let raft_node_id = raft_node_id.clone();

script/research/raft-diag/tmux_sessions.sh

@@ -1,12 +1,12 @@
 #!/bin/sh
-#export LOG_TARGETS='!net,!sled,!rustls' 
+export LOG_TARGETS='raft' 
 
-tmux new-session -d "./target/release/raft-diag --inbound tcp://127.0.0.1:12001 --path test1.db"
+tmux new-session -d "./target/release/raft-diag --inbound tcp://127.0.0.1:12001 --path test1.db -v"
 sleep 3
-tmux split-window -v "./target/release/raft-diag --inbound tcp://127.0.0.1:12002 --seeds tcp://127.0.0.1:12001 --outbound 3 --path test2.db"
+tmux split-window -v "./target/release/raft-diag --inbound tcp://127.0.0.1:12002 --seeds tcp://127.0.0.1:12001 --outbound 3 --path test2.db -v "
 sleep 2
-tmux split-window -h "./target/release/raft-diag  --seeds tcp://127.0.0.1:12001 --outbound 3 --path test3.db"
+tmux split-window -h "./target/release/raft-diag  --seeds tcp://127.0.0.1:12001 --outbound 3 --path test3.db -v"
 sleep 1
 tmux select-pane -t 0
-tmux split-window -h "./target/release/raft-diag  --seeds tcp://127.0.0.1:12001 --outbound 3 --path test4.db -b 3"
+tmux split-window -h "./target/release/raft-diag  --seeds tcp://127.0.0.1:12001 --outbound 3 --path test4.db -b 3 -v"
 tmux attach

src/raft/consensus.rs

@@ -2,15 +2,14 @@ use async_std::{
     sync::{Arc, Mutex},
     task,
 };
-use std::{cmp::min, path::PathBuf, time::Duration};
+use std::time::Duration;
 
 use async_executor::Executor;
 use chrono::Utc;
 use futures::{select, FutureExt};
 use fxhash::FxHashMap;
-use log::{debug, error, info, warn};
+use log::{debug, error, warn};
 use rand::{rngs::OsRng, Rng, RngCore};
-use url::Url;
 
 use crate::{
     net,
@@ -22,117 +21,37 @@ use crate::{
 };
 
 use super::{
+    gen_id, p2p_send_loop,
     primitives::{
         BroadcastMsgRequest, Channel, Log, LogRequest, LogResponse, Logs, MapLength, NetMsg,
-        NetMsgMethod, NodeId, Role, Sender, SyncRequest, SyncResponse, VoteRequest, VoteResponse,
+        NetMsgMethod, NodeId, NodeIdMsg, Role, Sender, VoteRequest, VoteResponse,
     },
-    DataStore,
+    prune_map, DataStore, RaftSettings,
 };
 
-#[derive(Clone, Debug)]
-pub struct RaftSettings {
-    // Milliseconds
-    pub heartbeat_timeout: u64,
-    pub timeout: u64,
-    pub load_ids_timeout: u64,
-    pub attempt_sync_timeout: u64,
-
-    // Seconds
-    pub prun_messages_duration: i64,
-    pub sync_attempts: u64,
-
-    // Datastore path
-    pub datastore_path: PathBuf,
-
-    // Self external address
-    pub external_addr: Option<Url>,
-}
-
-impl Default for RaftSettings {
-    fn default() -> Self {
-        Self {
-            heartbeat_timeout: 500,
-            timeout: 6000,
-            load_ids_timeout: 1000,
-            attempt_sync_timeout: 2000,
-            prun_messages_duration: 120,
-            sync_attempts: 60,
-            datastore_path: PathBuf::from(""),
-            external_addr: None,
-        }
-    }
-}
-
-async fn load_node_ids_loop(
-    nodes: Arc<Mutex<FxHashMap<NodeId, Url>>>,
-    p2p: net::P2pPtr,
-    role: Role,
-    self_addr: Url,
-    timeout: u64,
-) -> Result<()> {
-    if role == Role::Listener {
-        return Ok(())
-    }
-
+async fn send_node_id_loop(sender: async_channel::Sender<()>, timeout: i64) -> Result<()> {
     loop {
-        debug!(target: "raft", "Loading node ids from p2p hosts",);
-        task::sleep(Duration::from_millis(timeout)).await;
-        let hosts = p2p.hosts().clone();
-        let nodes_ip = hosts.load_all().await.clone();
-
-        for ip in nodes_ip.iter() {
-            if ip == &self_addr {
-                continue
-            }
-            (*nodes.lock().await).insert(NodeId::from(ip.clone()), ip.clone());
-        }
-    }
-}
-
-// Auxilary function to periodically prun seen messages, based on when they were received.
-// This helps us to prevent broadcasting loops.
-async fn prune_seen_messages(map: Arc<Mutex<fxhash::FxHashMap<String, i64>>>, seen_duration: i64) {
-    loop {
-        util::sleep(seen_duration as u64).await;
-        debug!("Pruning seen messages");
-
-        let now = Utc::now().timestamp();
-
-        let mut map = map.lock().await;
-        for (k, v) in map.clone().iter() {
-            if now - v > seen_duration {
-                map.remove(k);
-            }
-        }
-    }
-}
-
-async fn p2p_send_loop(receiver: async_channel::Receiver<NetMsg>, p2p: net::P2pPtr) -> Result<()> {
-    loop {
-        let msg: NetMsg = receiver.recv().await?;
-        if let Err(e) = p2p.broadcast(msg).await {
-            error!(target: "raft", "error occurred during broadcasting a msg: {}", e);
-            continue
-        }
+        util::sleep(timeout as u64).await;
+        sender.send(()).await?;
     }
 }
 
 pub struct Raft<T> {
     // this will be derived from the ip
-    pub id: Option<NodeId>,
+    id: NodeId,
 
-    role: Role,
+    pub(super) role: Role,
 
-    current_leader: Option<NodeId>,
+    pub(super) current_leader: NodeId,
 
-    votes_received: Vec<NodeId>,
+    pub(super) votes_received: Vec<NodeId>,
 
-    sent_length: MapLength,
-    acked_length: MapLength,
+    pub(super) sent_length: MapLength,
+    pub(super) acked_length: MapLength,
 
-    nodes: Arc<Mutex<FxHashMap<NodeId, Url>>>,
+    pub(super) nodes: Arc<Mutex<FxHashMap<NodeId, i64>>>,
 
-    last_term: u64,
+    pub(super) last_term: u64,
 
     sender: Sender,
 
@@ -164,13 +83,21 @@ impl<T: Decodable + Encodable + Clone> Raft<T> {
 
         let sender = async_channel::unbounded::<NetMsg>();
 
-        let id = settings.external_addr.clone().map(NodeId::from);
-        let role = if id.is_some() { Role::Follower } else { Role::Listener };
+        let id = match datastore.id.get_last()? {
+            Some(_id) => _id,
+            None => {
+                let id = gen_id();
+                datastore.id.insert(&id)?;
+                id
+            }
+        };
+
+        let role = Role::Follower;
 
         Ok(Self {
             id,
             role,
-            current_leader: None,
+            current_leader: NodeId("".into()),
             votes_received: vec![],
             sent_length: MapLength(FxHashMap::default()),
             acked_length: MapLength(FxHashMap::default()),
@@ -194,108 +121,65 @@ impl<T: Decodable + Encodable + Clone> Raft<T> {
     ) -> Result<()> {
         let p2p_send_task = executor.spawn(p2p_send_loop(self.sender.1.clone(), p2p.clone()));
 
-        if self.role != Role::Listener {
-            let self_addr = self.settings.external_addr.as_ref().unwrap().clone();
-            executor
-                .spawn(load_node_ids_loop(
-                    self.nodes.clone(),
-                    p2p.clone(),
-                    self.role.clone(),
-                    self_addr,
-                    self.settings.load_ids_timeout,
-                ))
-                .detach();
-        }
-
-        let prune_seen_messages_task = executor.spawn(prune_seen_messages(
+        let prune_seen_messages_task = executor.spawn(prune_map::<String>(
             self.seen_msgs.clone(),
             self.settings.prun_messages_duration,
         ));
 
-        let mut synced = true;
-
-        // Sync listener node
-        if self.role == Role::Listener {
-            synced = false;
-            let last_term = if !self.is_logs_empty() { self.last_log()?.unwrap().term } else { 0 };
-
-            let sync_request_id = OsRng.next_u64();
-            let sync_request =
-                SyncRequest { id: sync_request_id, logs_len: self.logs_len(), last_term };
-
-            info!("Start Syncing...");
-            for _ in 0..self.settings.sync_attempts {
-                if synced {
-                    break
-                }
-
-                self.send(None, &serialize(&sync_request), NetMsgMethod::SyncRequest, None).await?;
+        let prune_nodes_id_task = executor
+            .spawn(prune_map::<NodeId>(self.nodes.clone(), self.settings.prun_nodes_ids_duration));
 
-                synced = self
-                    .waiting_for_sync(
-                        executor.clone(),
-                        p2p_recv_channel.clone(),
-                        stop_signal.clone(),
-                        sync_request_id,
-                    )
-                    .await?;
-            }
-            if synced {
-                info!("SYNCED SUCCESSFULLY!!");
-            }
-        }
+        let (node_id_sx, node_id_rv) = async_channel::unbounded::<()>();
+        let send_node_id_loop_task =
+            executor.spawn(send_node_id_loop(node_id_sx, self.settings.node_id_timeout));
 
         let mut rng = rand::thread_rng();
 
         let broadcast_msg_rv = self.msgs_channel.1.clone();
 
-        if !synced {
-            error!("SYNCING FAILED!!");
-        } else {
-            loop {
-                let timeout: Duration = if self.role == Role::Leader {
-                    Duration::from_millis(self.settings.heartbeat_timeout)
-                } else {
-                    Duration::from_millis(
-                        rng.gen_range(0..self.settings.heartbeat_timeout) + self.settings.timeout,
-                    )
-                };
-
-                let result: Result<()>;
-
-                select! {
-                    m =  p2p_recv_channel.recv().fuse() => result = self.handle_method(m?).await,
-                    m =  broadcast_msg_rv.recv().fuse() => result = self.broadcast_msg(&m?,None).await,
-                    _ = task::sleep(timeout).fuse() => {
-                        result = if self.role == Role::Leader {
-                            self.send_heartbeat().await
-                        }else {
-                            self.send_vote_request().await
-                        };
-                    },
-                    _ = stop_signal.recv().fuse() => break,
-                }
+        loop {
+            let timeout = if self.role == Role::Leader {
+                self.settings.heartbeat_timeout
+            } else {
+                rng.gen_range(0..self.settings.timeout) + self.settings.timeout
+            };
+            let timeout = Duration::from_millis(timeout);
 
-                match result {
-                    Ok(_) => {}
-                    Err(e) => warn!(target: "raft", "warn: {}", e),
-                }
+            let result: Result<()>;
+
+            select! {
+                m =  p2p_recv_channel.recv().fuse() => result = self.handle_method(m?).await,
+                m =  broadcast_msg_rv.recv().fuse() => result = self.broadcast_msg(&m?,None).await,
+                _ =  node_id_rv.recv().fuse() => result = self.send_node_id_msg().await,
+                _ = task::sleep(timeout).fuse() => {
+                    result = if self.role == Role::Leader {
+                        self.send_heartbeat().await
+                    }else {
+                        self.send_vote_request().await
+                    };
+                },
+                _ = stop_signal.recv().fuse() => break,
+            }
+
+            match result {
+                Ok(_) => {}
+                Err(e) => warn!(target: "raft", "warn: {}", e),
             }
         }
 
         warn!(target: "raft", "Raft Terminating...");
         p2p_send_task.cancel().await;
         prune_seen_messages_task.cancel().await;
+        prune_nodes_id_task.cancel().await;
+        send_node_id_loop_task.cancel().await;
         self.datastore.flush().await?;
         Ok(())
     }
 
-    pub fn get_commits_channel(&self) -> async_channel::Receiver<T> {
-        self.commits_channel.1.clone()
-    }
-
-    pub fn get_msgs_channel(&self) -> async_channel::Sender<T> {
-        self.msgs_channel.0.clone()
+    async fn send_node_id_msg(&self) -> Result<()> {
+        let node_id_msg = serialize(&NodeIdMsg { id: self.id.clone() });
+        self.send(None, &node_id_msg, NetMsgMethod::NodeIdMsg, None).await?;
+        Ok(())
     }
 
     async fn broadcast_msg(&mut self, msg: &T, msg_id: Option<u64>) -> Result<()> {
@@ -304,11 +188,11 @@ impl<T: Decodable + Encodable + Clone> Raft<T> {
             let log = Log { msg, term: self.current_term()? };
             self.push_log(&log)?;
 
-            self.acked_length.insert(&self.id.clone().unwrap(), self.logs_len());
+            self.acked_length.insert(&self.id, self.logs_len());
         } else {
             let b_msg = BroadcastMsgRequest(serialize(msg));
             self.send(
-                self.current_leader.clone(),
+                Some(self.current_leader.clone()),
                 &serialize(&b_msg),
                 NetMsgMethod::BroadcastRequest,
                 msg_id,
@@ -316,7 +200,7 @@ impl<T: Decodable + Encodable + Clone> Raft<T> {
             .await?;
         }
 
-        debug!(target: "raft", "Role: {:?}, broadcast a msg id: {:?} ", self.role, msg_id);
+        debug!(target: "raft", "Role: {:?} Id: {:?}, broadcast a msg id: {:?} ", self.role, self.id, msg_id);
 
         Ok(())
     }
@@ -344,81 +228,20 @@ impl<T: Decodable + Encodable + Clone> Raft<T> {
                 let d: T = deserialize(&vr.0)?;
                 self.broadcast_msg(&d, Some(msg.id)).await?;
             }
-            NetMsgMethod::SyncRequest => {
-                debug!(target: "raft", "Receive sync request");
-                let sr: SyncRequest = deserialize(&msg.payload)?;
-                self.receive_sync_request(&sr, msg.id).await?;
-            }
-            NetMsgMethod::SyncResponse => {}
-        }
-
-        debug!(target: "raft", "Role: {:?}  receive a msg with id: {}  recipient_id: {:?} method: {:?} ",
-               self.role, msg.id, &msg.recipient_id.is_some(), &msg.method);
-        Ok(())
-    }
-
-    async fn receive_sync_request(&self, sr: &SyncRequest, msg_id: u64) -> Result<()> {
-        if self.role == Role::Leader {
-            let mut wipe = false;
-
-            let logs = if sr.logs_len == 0 {
-                self.logs()?.clone()
-            } else if self.logs_len() >= sr.logs_len &&
-                self.get_log(sr.logs_len - 1)?.term == sr.last_term
-            {
-                self.slice_logs_from(sr.logs_len)?.unwrap()
-            } else {
-                wipe = true;
-                self.logs()?.clone()
-            };
-
-            let sync_response = SyncResponse {
-                id: sr.id,
-                logs,
-                commit_length: self.commits_len(),
-                leader_id: self.id.clone().unwrap(),
-                wipe,
-            };
-
-            debug!(target: "raft", "Send sync response");
-            self.send(None, &serialize(&sync_response), NetMsgMethod::SyncResponse, None).await?;
-        } else {
-            self.send(
-                self.current_leader.clone(),
-                &serialize(sr),
-                NetMsgMethod::SyncRequest,
-                Some(msg_id),
-            )
-            .await?;
-        }
-
-        Ok(())
-    }
-
-    async fn receive_sync_response(&mut self, sr: &SyncResponse) -> Result<()> {
-        debug!(target: "raft", "Receive sync response");
-        if sr.wipe {
-            self.push_logs(&sr.logs)?;
-        } else {
-            for log in sr.logs.0.iter() {
-                self.push_log(log)?;
+            NetMsgMethod::NodeIdMsg => {
+                let node_id_msg: NodeIdMsg = deserialize(&msg.payload)?;
+                if node_id_msg.id != self.id {
+                    self.nodes.lock().await.insert(node_id_msg.id, Utc::now().timestamp());
+                }
             }
         }
 
-        if !self.logs()?.is_empty() {
-            self.set_current_term(&self.logs()?.0.last().unwrap().term.clone())?;
-        }
-
-        for i in self.commits_len()..sr.commit_length {
-            self.push_commit(&self.get_log(i)?.msg).await?;
-        }
-
-        self.current_leader = Some(sr.leader_id.clone());
-
+        debug!(target: "raft", "Role: {:?} Id: {:?}, receive a msg with id: {}  recipient_id: {:?} method: {:?} ",
+               self.role, self.id, msg.id, &msg.recipient_id, &msg.method);
         Ok(())
     }
 
-    async fn send(
+    pub(super) async fn send(
         &self,
         recipient_id: Option<NodeId>,
         payload: &[u8],
@@ -427,8 +250,8 @@ impl<T: Decodable + Encodable + Clone> Raft<T> {
     ) -> Result<()> {
         let random_id = if msg_id.is_some() { msg_id.unwrap() } else { OsRng.next_u64() };
 
-        debug!(target: "raft","Role: {:?}  send a msg with id: {}  recipient_id: {:?} method: {:?} ",
-               self.role, random_id, &recipient_id.is_some(), &method);
+        debug!(target: "raft","Role: {:?} Id: {:?}, send a msg with id: {}  recipient_id: {:?} method: {:?} ",
+               self.role, self.id, random_id, &recipient_id, &method);
 
         let net_msg = NetMsg { id: random_id, recipient_id, payload: payload.to_vec(), method };
         self.seen_msgs.lock().await.insert(random_id.to_string(), Utc::now().timestamp());
@@ -437,253 +260,7 @@ impl<T: Decodable + Encodable + Clone> Raft<T> {
         Ok(())
     }
 
-    async fn waiting_for_sync(
-        &mut self,
-        executor: Arc<Executor<'_>>,
-        p2p_recv_channel: async_channel::Receiver<NetMsg>,
-        stop_signal: async_channel::Receiver<()>,
-        sync_request_id: u64,
-    ) -> Result<bool> {
-        let (timeout_s, timeout_r) = async_channel::unbounded::<()>();
-        let attempt_sync_timeout = self.settings.attempt_sync_timeout;
-        executor
-            .spawn(async move {
-                task::sleep(Duration::from_millis(attempt_sync_timeout)).await;
-                timeout_s.send(()).await.unwrap_or(());
-            })
-            .detach();
-
-        loop {
-            select! {
-                msg =  p2p_recv_channel.recv().fuse() => {
-                    let msg = msg?;
-                    if msg.method != NetMsgMethod::SyncResponse {
-                        continue
-                    }
-
-                    let sr: SyncResponse = deserialize(&msg.payload)?;
-                    if sr.id != sync_request_id {
-                        continue
-                    }
-
-                    self.receive_sync_response(&sr).await?;
-                    return Ok(true)
-                },
-                _ = stop_signal.recv().fuse() => break,
-                _ = timeout_r.recv().fuse() => break,
-            }
-        }
-        Ok(false)
-    }
-
-    async fn send_heartbeat(&self) -> Result<()> {
-        if self.role == Role::Leader {
-            let nodes = self.nodes.lock().await;
-            let nodes_cloned = nodes.clone();
-            drop(nodes);
-            for node in nodes_cloned.iter() {
-                self.update_logs(node.0).await?;
-            }
-        }
-        Ok(())
-    }
-
-    async fn send_vote_request(&mut self) -> Result<()> {
-        if self.role == Role::Listener {
-            return Ok(())
-        }
-
-        let self_id = self.id.clone().unwrap();
-
-        self.set_current_term(&(self.current_term()? + 1))?;
-        self.role = Role::Candidate;
-        self.set_voted_for(&Some(self_id.clone()))?;
-        self.votes_received = vec![];
-        self.votes_received.push(self_id.clone());
-
-        self.reset_last_term()?;
-
-        let request = VoteRequest {
-            node_id: self_id,
-            current_term: self.current_term()?,
-            log_length: self.logs_len(),
-            last_term: self.last_term,
-        };
-
-        let payload = serialize(&request);
-        self.send(None, &payload, NetMsgMethod::VoteRequest, None).await
-    }
-
-    async fn receive_vote_request(&mut self, vr: VoteRequest) -> Result<()> {
-        if self.role == Role::Listener {
-            return Ok(())
-        }
-
-        if vr.current_term > self.current_term()? {
-            self.set_current_term(&vr.current_term)?;
-            self.set_voted_for(&None)?;
-            self.role = Role::Follower;
-        }
-
-        self.reset_last_term()?;
-
-        // check the logs of the candidate
-        let vote_ok = (vr.last_term > self.last_term) ||
-            (vr.last_term == self.last_term && vr.log_length >= self.logs_len());
-
-        // slef.voted_for equal to vr.node_id or is None or voted to someone else
-        let vote =
-            if let Some(voted_for) = self.voted_for()? { voted_for == vr.node_id } else { true };
-
-        let mut response = VoteResponse {
-            node_id: self.id.clone().unwrap(),
-            current_term: self.current_term()?,
-            ok: false,
-        };
-
-        if vr.current_term == self.current_term()? && vote_ok && vote {
-            self.set_voted_for(&Some(vr.node_id.clone()))?;
-            response.set_ok(true);
-        }
-
-        let payload = serialize(&response);
-        self.send(Some(vr.node_id), &payload, NetMsgMethod::VoteResponse, None).await
-    }
-
-    async fn receive_vote_response(&mut self, vr: VoteResponse) -> Result<()> {
-        if self.role == Role::Listener {
-            return Ok(())
-        }
-
-        if self.role == Role::Candidate && vr.current_term == self.current_term()? && vr.ok {
-            self.votes_received.push(vr.node_id);
-
-            let nodes = self.nodes.lock().await;
-            let nodes_cloned = nodes.clone();
-            drop(nodes);
-
-            if self.votes_received.len() >= (nodes_cloned.len() / 2) {
-                self.role = Role::Leader;
-                self.current_leader = Some(self.id.clone().unwrap());
-                for node in nodes_cloned.iter() {
-                    self.sent_length.insert(node.0, self.logs_len());
-                    self.acked_length.insert(node.0, 0);
-                }
-            }
-        } else if vr.current_term > self.current_term()? {
-            self.set_current_term(&vr.current_term)?;
-            self.role = Role::Follower;
-            self.set_voted_for(&None)?;
-        }
-
-        Ok(())
-    }
-
-    // only the leader broadcast this
-    async fn update_logs(&self, node_id: &NodeId) -> Result<()> {
-        let prefix_len = match self.sent_length.get(node_id) {
-            Ok(len) => len,
-            Err(_) => {
-                // return if failed to index
-                return Ok(())
-            }
-        };
-
-        let suffix: Logs = match self.slice_logs_from(prefix_len)? {
-            Some(l) => l,
-            None => return Ok(()),
-        };
-
-        let mut prefix_term = 0;
-
-        if prefix_len > 0 {
-            prefix_term = self.get_log(prefix_len - 1)?.term;
-        }
-
-        let request = LogRequest {
-            leader_id: self.id.clone().unwrap(),
-            current_term: self.current_term()?,
-            prefix_len,
-            prefix_term,
-            commit_length: self.commits_len(),
-            suffix,
-        };
-
-        let payload = serialize(&request);
-        self.send(Some(node_id.clone()), &payload, NetMsgMethod::LogRequest, None).await
-    }
-
-    async fn receive_log_request(&mut self, lr: LogRequest) -> Result<()> {
-        debug!(target: "raft",
-        "Receive LogRequest current_term: {} prefix_term: {} prefix_len: {} commit_length: {} suffixlen {}",
-        lr.current_term, lr.prefix_term, lr.prefix_len, lr.commit_length, lr.suffix.len(),
-        );
-
-        if lr.current_term > self.current_term()? {
-            self.set_current_term(&lr.current_term)?;
-            self.set_voted_for(&None)?;
-        }
-
-        if lr.current_term == self.current_term()? {
-            if self.role != Role::Listener {
-                self.role = Role::Follower;
-            }
-            self.current_leader = Some(lr.leader_id.clone());
-        }
-
-        let mut ok = (self.logs_len() >= lr.prefix_len) &&
-            (lr.prefix_len == 0 || self.get_log(lr.prefix_len - 1)?.term == lr.prefix_term);
-
-        let mut ack = 0;
-
-        if lr.current_term == self.current_term()? && ok {
-            self.append_log(lr.prefix_len, lr.commit_length, &lr.suffix).await?;
-            ack = lr.prefix_len + lr.suffix.len();
-        } else {
-            ok = false;
-        }
-
-        if self.role == Role::Listener {
-            return Ok(())
-        }
-
-        let response = LogResponse {
-            node_id: self.id.clone().unwrap(),
-            current_term: self.current_term()?,
-            ack,
-            ok,
-        };
-
-        debug!(target: "raft",
-         "Send LogResponse current_term: {} ack: {} ok: {}",
-         response.current_term, response.ack, response.ok
-        );
-
-        let payload = serialize(&response);
-        self.send(Some(lr.leader_id.clone()), &payload, NetMsgMethod::LogResponse, None).await
-    }
-
-    async fn receive_log_response(&mut self, lr: LogResponse) -> Result<()> {
-        if lr.current_term == self.current_term()? && self.role == Role::Leader {
-            if lr.ok && lr.ack >= self.acked_length.get(&lr.node_id)? {
-                self.sent_length.insert(&lr.node_id, lr.ack);
-                self.acked_length.insert(&lr.node_id, lr.ack);
-                self.commit_log().await?;
-            } else if self.sent_length.get(&lr.node_id)? > 0 {
-                self.sent_length.insert(&lr.node_id, self.sent_length.get(&lr.node_id)? - 1);
-            }
-        } else if lr.current_term > self.current_term()? {
-            self.set_current_term(&lr.current_term)?;
-            if self.role != Role::Listener {
-                self.role = Role::Follower;
-            }
-            self.set_voted_for(&None)?;
-        }
-
-        Ok(())
-    }
-
-    fn reset_last_term(&mut self) -> Result<()> {
+    pub(super) fn reset_last_term(&mut self) -> Result<()> {
         self.last_term = 0;
 
         if let Some(log) = self.last_log()? {
@@ -693,122 +270,74 @@ impl<T: Decodable + Encodable + Clone> Raft<T> {
         Ok(())
     }
 
-    fn acks(&self, nodes: FxHashMap<NodeId, Url>, length: u64) -> FxHashMap<NodeId, Url> {
-        nodes
-            .into_iter()
-            .filter(|n| {
-                let len = self.acked_length.get(&n.0);
-                len.is_ok() && len.unwrap() >= length
-            })
-            .collect()
-    }
-
-    async fn commit_log(&mut self) -> Result<()> {
-        let nodes_ptr = self.nodes.lock().await;
-        let min_acks = ((nodes_ptr.len() + 1) / 2) as usize;
-        let nodes = nodes_ptr.clone();
-        drop(nodes_ptr);
-
-        let mut ready: Vec<u64> = vec![];
-
-        for len in 1..(self.logs_len() + 1) {
-            if self.acks(nodes.clone(), len).len() >= min_acks {
-                ready.push(len);
-            }
-        }
-
-        if ready.is_empty() {
-            return Ok(())
-        }
-
-        let max_ready = *ready.iter().max().unwrap();
-
-        if max_ready > self.commits_len() &&
-            self.get_log(max_ready - 1)?.term == self.current_term()?
-        {
-            for i in self.commits_len()..max_ready {
-                self.push_commit(&self.get_log(i)?.msg).await?;
-            }
-        }
-
-        Ok(())
-    }
-
-    async fn append_log(
-        &mut self,
-        prefix_len: u64,
-        leader_commit: u64,
-        suffix: &Logs,
-    ) -> Result<()> {
-        if !suffix.is_empty() && self.logs_len() > prefix_len {
-            let index = min(self.logs_len(), prefix_len + suffix.len()) - 1;
-            if self.get_log(index)?.term != suffix.get(index - prefix_len)?.term {
-                self.push_logs(&self.slice_logs_to(prefix_len)?)?;
-            }
-        }
-
-        if prefix_len + suffix.len() > self.logs_len() {
-            for i in (self.logs_len() - prefix_len)..suffix.len() {
-                self.push_log(&suffix.get(i)?)?;
-            }
-        }
-
-        if leader_commit > self.commits_len() {
-            for i in self.commits_len()..leader_commit {
-                self.push_commit(&self.get_log(i)?.msg).await?;
-            }
-        }
-
-        Ok(())
-    }
-
-    fn set_current_term(&mut self, i: &u64) -> Result<()> {
+    pub(super) fn set_current_term(&mut self, i: &u64) -> Result<()> {
         self.datastore.current_term.insert(i)
     }
-    fn set_voted_for(&mut self, i: &Option<NodeId>) -> Result<()> {
+
+    pub(super) fn set_voted_for(&mut self, i: &Option<NodeId>) -> Result<()> {
         self.datastore.voted_for.insert(i)
     }
-    async fn push_commit(&mut self, commit: &[u8]) -> Result<()> {
+
+    pub(super) async fn push_commit(&mut self, commit: &[u8]) -> Result<()> {
         let commit: T = deserialize(commit)?;
         self.commits_channel.0.send(commit.clone()).await?;
         self.datastore.commits.insert(&commit)
     }
-    fn push_log(&mut self, log: &Log) -> Result<()> {
+
+    pub(super) fn push_log(&mut self, log: &Log) -> Result<()> {
         self.datastore.logs.insert(log)
     }
-    fn push_logs(&mut self, logs: &Logs) -> Result<()> {
+
+    pub(super) fn push_logs(&mut self, logs: &Logs) -> Result<()> {
         self.datastore.logs.wipe_insert_all(&logs.to_vec())
     }
-    fn is_logs_empty(&self) -> bool {
-        self.datastore.logs.is_empty()
-    }
 
-    fn current_term(&self) -> Result<u64> {
+    pub(super) fn current_term(&self) -> Result<u64> {
         Ok(self.datastore.current_term.get_last()?.unwrap_or(0))
     }
-    fn voted_for(&self) -> Result<Option<NodeId>> {
+
+    pub(super) fn voted_for(&self) -> Result<Option<NodeId>> {
         Ok(self.datastore.voted_for.get_last()?.flatten())
     }
-    fn commits_len(&self) -> u64 {
+
+    pub fn get_commits_channel(&self) -> async_channel::Receiver<T> {
+        self.commits_channel.1.clone()
+    }
+
+    pub fn get_msgs_channel(&self) -> async_channel::Sender<T> {
+        self.msgs_channel.0.clone()
+    }
+
+    pub fn get_id(&self) -> NodeId {
+        self.id.clone()
+    }
+
+    pub(super) fn commits_len(&self) -> u64 {
         self.datastore.commits.len()
     }
+
     fn logs(&self) -> Result<Logs> {
         Ok(Logs(self.datastore.logs.get_all()?))
     }
-    fn logs_len(&self) -> u64 {
+
+    pub(super) fn logs_len(&self) -> u64 {
         self.datastore.logs.len()
     }
+
     fn last_log(&self) -> Result<Option<Log>> {
         self.datastore.logs.get_last()
     }
-    fn get_log(&self, index: u64) -> Result<Log> {
+
+    pub(super) fn get_log(&self, index: u64) -> Result<Log> {
         self.datastore.logs.get(index)
     }
-    fn slice_logs_from(&self, index: u64) -> Result<Option<Logs>> {
+
+    pub(super) fn slice_logs_from(&self, index: u64) -> Result<Option<Logs>> {
         let logs = self.logs()?;
         Ok(logs.slice_from(index))
     }
-    fn slice_logs_to(&self, index: u64) -> Result<Logs> {
+
+    pub(super) fn slice_logs_to(&self, index: u64) -> Result<Logs> {
         let logs = self.logs()?;
         Ok(logs.slice_to(index))
     }

src/raft/consensus_candidate.rs

@@ -0,0 +1,58 @@
+use crate::{
+    util::serial::{serialize, Decodable, Encodable},
+    Result,
+};
+
+use super::{
+    primitives::{NetMsgMethod, Role, VoteRequest, VoteResponse},
+    Raft,
+};
+
+impl<T: Decodable + Encodable + Clone> Raft<T> {
+    pub(super) async fn send_vote_request(&mut self) -> Result<()> {
+        let self_id = self.get_id();
+
+        self.set_current_term(&(self.current_term()? + 1))?;
+        self.role = Role::Candidate;
+        self.set_voted_for(&Some(self_id.clone()))?;
+        self.votes_received = vec![];
+        self.votes_received.push(self_id.clone());
+
+        self.reset_last_term()?;
+
+        let request = VoteRequest {
+            node_id: self_id,
+            current_term: self.current_term()?,
+            log_length: self.logs_len(),
+            last_term: self.last_term,
+        };
+
+        let payload = serialize(&request);
+        self.send(None, &payload, NetMsgMethod::VoteRequest, None).await
+    }
+
+    pub(super) async fn receive_vote_response(&mut self, vr: VoteResponse) -> Result<()> {
+        if self.role == Role::Candidate && vr.current_term == self.current_term()? && vr.ok {
+            self.votes_received.push(vr.node_id);
+
+            let nodes = self.nodes.lock().await;
+            let nodes_cloned = nodes.clone();
+            drop(nodes);
+
+            if self.votes_received.len() >= ((nodes_cloned.len() + 1) / 2) {
+                self.role = Role::Leader;
+                self.current_leader = self.get_id();
+                for node in nodes_cloned.iter() {
+                    self.sent_length.insert(node.0, self.logs_len());
+                    self.acked_length.insert(node.0, 0);
+                }
+            }
+        } else if vr.current_term > self.current_term()? {
+            self.set_current_term(&vr.current_term)?;
+            self.role = Role::Follower;
+            self.set_voted_for(&None)?;
+        }
+
+        Ok(())
+    }
+}

src/raft/consensus_follower.rs

@@ -0,0 +1,112 @@
+use std::cmp::min;
+
+use log::debug;
+
+use crate::{
+    util::serial::{serialize, Decodable, Encodable},
+    Result,
+};
+
+use super::{
+    primitives::{LogRequest, LogResponse, Logs, NetMsgMethod, Role, VoteRequest, VoteResponse},
+    Raft,
+};
+
+impl<T: Decodable + Encodable + Clone> Raft<T> {
+    pub(super) async fn receive_vote_request(&mut self, vr: VoteRequest) -> Result<()> {
+        if vr.current_term > self.current_term()? {
+            self.set_current_term(&vr.current_term)?;
+            self.set_voted_for(&None)?;
+            self.role = Role::Follower;
+        }
+
+        self.reset_last_term()?;
+
+        // check the logs of the candidate
+        let vote_ok = (vr.last_term > self.last_term) ||
+            (vr.last_term == self.last_term && vr.log_length >= self.logs_len());
+
+        // slef.voted_for equal to vr.node_id or is None or voted to someone else
+        let vote =
+            if let Some(voted_for) = self.voted_for()? { voted_for == vr.node_id } else { true };
+
+        let mut response =
+            VoteResponse { node_id: self.get_id(), current_term: self.current_term()?, ok: false };
+
+        if vr.current_term == self.current_term()? && vote_ok && vote {
+            self.set_voted_for(&Some(vr.node_id.clone()))?;
+            response.set_ok(true);
+        }
+
+        let payload = serialize(&response);
+        self.send(Some(vr.node_id), &payload, NetMsgMethod::VoteResponse, None).await
+    }
+
+    pub(super) async fn receive_log_request(&mut self, lr: LogRequest) -> Result<()> {
+        debug!(target: "raft",
+        "Receive LogRequest current_term: {} prefix_term: {} prefix_len: {} commit_length: {} suffixlen {}",
+        lr.current_term, lr.prefix_term, lr.prefix_len, lr.commit_length, lr.suffix.len(),
+        );
+
+        if lr.current_term > self.current_term()? {
+            self.set_current_term(&lr.current_term)?;
+            self.set_voted_for(&None)?;
+        }
+
+        if lr.current_term == self.current_term()? {
+            self.role = Role::Follower;
+            self.current_leader = lr.leader_id.clone();
+        }
+
+        let mut ok = (self.logs_len() >= lr.prefix_len) &&
+            (lr.prefix_len == 0 || self.get_log(lr.prefix_len - 1)?.term == lr.prefix_term);
+
+        let mut ack = 0;
+
+        if lr.current_term == self.current_term()? && ok {
+            self.append_log(lr.prefix_len, lr.commit_length, &lr.suffix).await?;
+            ack = lr.prefix_len + lr.suffix.len();
+        } else {
+            ok = false;
+        }
+
+        let response =
+            LogResponse { node_id: self.get_id(), current_term: self.current_term()?, ack, ok };
+
+        debug!(target: "raft",
+         "Send LogResponse current_term: {} ack: {} ok: {}",
+         response.current_term, response.ack, response.ok
+        );
+
+        let payload = serialize(&response);
+        self.send(Some(lr.leader_id.clone()), &payload, NetMsgMethod::LogResponse, None).await
+    }
+
+    async fn append_log(
+        &mut self,
+        prefix_len: u64,
+        leader_commit: u64,
+        suffix: &Logs,
+    ) -> Result<()> {
+        if !suffix.is_empty() && self.logs_len() > prefix_len {
+            let index = min(self.logs_len(), prefix_len + suffix.len()) - 1;
+            if self.get_log(index)?.term != suffix.get(index - prefix_len)?.term {
+                self.push_logs(&self.slice_logs_to(prefix_len)?)?;
+            }
+        }
+
+        if prefix_len + suffix.len() > self.logs_len() {
+            for i in (self.logs_len() - prefix_len)..suffix.len() {
+                self.push_log(&suffix.get(i)?)?;
+            }
+        }
+
+        if leader_commit > self.commits_len() {
+            for i in self.commits_len()..leader_commit {
+                self.push_commit(&self.get_log(i)?.msg).await?;
+            }
+        }
+
+        Ok(())
+    }
+}

src/raft/consensus_leader.rs

@@ -0,0 +1,116 @@
+use fxhash::FxHashMap;
+
+use crate::{
+    util::serial::{serialize, Decodable, Encodable},
+    Result,
+};
+
+use super::{
+    primitives::{LogRequest, LogResponse, Logs, NetMsgMethod, NodeId, Role},
+    Raft,
+};
+
+impl<T: Decodable + Encodable + Clone> Raft<T> {
+    pub(super) async fn send_heartbeat(&mut self) -> Result<()> {
+        let nodes = self.nodes.lock().await;
+        let nodes_cloned = nodes.clone();
+        drop(nodes);
+        for node in nodes_cloned.iter() {
+            self.update_logs(node.0).await?;
+        }
+        Ok(())
+    }
+
+    async fn update_logs(&mut self, node_id: &NodeId) -> Result<()> {
+        let prefix_len = match self.sent_length.get(node_id) {
+            Ok(len) => len,
+            Err(_) => {
+                self.sent_length.insert(node_id, 0);
+                self.acked_length.insert(node_id, 0);
+                0
+            }
+        };
+
+        let suffix: Logs = match self.slice_logs_from(prefix_len)? {
+            Some(l) => l,
+            None => return Ok(()),
+        };
+
+        let mut prefix_term = 0;
+
+        if prefix_len > 0 {
+            prefix_term = self.get_log(prefix_len - 1)?.term;
+        }
+
+        let request = LogRequest {
+            leader_id: self.get_id(),
+            current_term: self.current_term()?,
+            prefix_len,
+            prefix_term,
+            commit_length: self.commits_len(),
+            suffix,
+        };
+
+        let payload = serialize(&request);
+        self.send(Some(node_id.clone()), &payload, NetMsgMethod::LogRequest, None).await
+    }
+
+    pub(super) async fn receive_log_response(&mut self, lr: LogResponse) -> Result<()> {
+        if lr.current_term == self.current_term()? && self.role == Role::Leader {
+            if lr.ok && lr.ack >= self.acked_length.get(&lr.node_id)? {
+                self.sent_length.insert(&lr.node_id, lr.ack);
+                self.acked_length.insert(&lr.node_id, lr.ack);
+                self.commit_log().await?;
+            } else if self.sent_length.get(&lr.node_id)? > 0 {
+                self.sent_length.insert(&lr.node_id, self.sent_length.get(&lr.node_id)? - 1);
+            }
+        } else if lr.current_term > self.current_term()? {
+            self.set_current_term(&lr.current_term)?;
+            self.role = Role::Follower;
+            self.set_voted_for(&None)?;
+        }
+
+        Ok(())
+    }
+
+    fn acks(&self, nodes: FxHashMap<NodeId, i64>, length: u64) -> FxHashMap<NodeId, i64> {
+        nodes
+            .into_iter()
+            .filter(|n| {
+                let len = self.acked_length.get(&n.0);
+                len.is_ok() && len.unwrap() >= length
+            })
+            .collect()
+    }
+
+    async fn commit_log(&mut self) -> Result<()> {
+        let nodes_ptr = self.nodes.lock().await;
+        let min_acks = ((nodes_ptr.len() + 1) / 2) as usize;
+        let nodes = nodes_ptr.clone();
+        drop(nodes_ptr);
+
+        let mut ready: Vec<u64> = vec![];
+
+        for len in 1..(self.logs_len() + 1) {
+            if self.acks(nodes.clone(), len).len() >= min_acks {
+                ready.push(len);
+            }
+        }
+
+        if ready.is_empty() {
+            return Ok(())
+        }
+
+        let max_ready = *ready.iter().max().unwrap();
+
+        if max_ready > self.commits_len() &&
+            self.get_log(max_ready - 1)?.term == self.current_term()?
+        {
+            for i in self.commits_len()..max_ready {
+                self.push_commit(&self.get_log(i)?.msg).await?;
+            }
+        }
+
+        Ok(())
+    }
+}

src/raft/datastore.rs

@@ -15,6 +15,7 @@ const SLED_COMMITS_TREE: &[u8] = b"_commits";
 const _SLED_COMMITS_LENGTH_TREE: &[u8] = b"_commit_length";
 const SLED_VOTED_FOR_TREE: &[u8] = b"_voted_for";
 const SLED_CURRENT_TERM_TREE: &[u8] = b"_current_term";
+const SLED_ID_TREE: &[u8] = b"_id";
 
 pub struct DataStore<T> {
     _db: sled::Db,
@@ -22,6 +23,7 @@ pub struct DataStore<T> {
     pub commits: DataTree<T>,
     pub voted_for: DataTree<Option<NodeId>>,
     pub current_term: DataTree<u64>,
+    pub id: DataTree<NodeId>,
 }
 
 impl<T: Encodable + Decodable> DataStore<T> {
@@ -31,8 +33,9 @@ impl<T: Encodable + Decodable> DataStore<T> {
         let commits = DataTree::new(&_db, SLED_COMMITS_TREE)?;
         let voted_for = DataTree::new(&_db, SLED_VOTED_FOR_TREE)?;
         let current_term = DataTree::new(&_db, SLED_CURRENT_TERM_TREE)?;
+        let id = DataTree::new(&_db, SLED_ID_TREE)?;
 
-        Ok(Self { _db, logs, commits, voted_for, current_term })
+        Ok(Self { _db, logs, commits, voted_for, current_term, id })
     }
     pub async fn flush(&self) -> Result<()> {
         debug!(target: "raft", "DataStore flush");

src/raft/mod.rs

@@ -1,9 +1,59 @@
+use async_std::sync::{Arc, Mutex};
+
+use chrono::Utc;
+use log::{debug, error};
+
+use crate::{net, util, Result};
+
 mod consensus;
+mod consensus_candidate;
+mod consensus_follower;
+mod consensus_leader;
 mod datastore;
 mod primitives;
 mod protocol_raft;
+mod settings;
 
-pub use consensus::{Raft, RaftSettings};
+pub use consensus::Raft;
 pub use datastore::DataStore;
 pub use primitives::NetMsg;
 pub use protocol_raft::ProtocolRaft;
+pub use settings::RaftSettings;
+
+use primitives::NodeId;
+
+// Auxilary function to periodically prun items, based on when they were received.
+async fn prune_map<T: Clone + Eq + std::hash::Hash>(
+    map: Arc<Mutex<fxhash::FxHashMap<T, i64>>>,
+    seen_duration: i64,
+) {
+    loop {
+        util::sleep(seen_duration as u64).await;
+        debug!(target: "raft", "Pruning item in map");
+
+        let now = Utc::now().timestamp();
+
+        let mut map = map.lock().await;
+        for (k, v) in map.clone().iter() {
+            if now - v > seen_duration {
+                map.remove(k);
+            }
+        }
+    }
+}
+
+async fn p2p_send_loop(receiver: async_channel::Receiver<NetMsg>, p2p: net::P2pPtr) -> Result<()> {
+    loop {
+        let msg: NetMsg = receiver.recv().await?;
+        if let Err(e) = p2p.broadcast(msg).await {
+            error!(target: "raft", "error occurred during broadcasting a msg: {}", e);
+            continue
+        }
+    }
+}
+
+fn gen_id() -> NodeId {
+    let timestamp = Utc::now().timestamp();
+    let hash: String = blake3::hash(&timestamp.to_be_bytes()).to_hex().to_string();
+    NodeId(hash)
+}

src/raft/primitives.rs

@@ -1,11 +1,10 @@
 use std::io;
 
 use fxhash::FxHashMap;
-use url::Url;
 
 use crate::{
     impl_vec,
-    util::serial::{serialize, Decodable, Encodable, SerialDecodable, SerialEncodable, VarInt},
+    util::serial::{Decodable, Encodable, SerialDecodable, SerialEncodable, VarInt},
     Error, Result,
 };
 
@@ -14,7 +13,6 @@ pub type Sender = (async_channel::Sender<NetMsg>, async_channel::Receiver<NetMsg
 
 #[derive(PartialEq, Eq, Debug, Clone)]
 pub enum Role {
-    Listener,
     Follower,
     Candidate,
     Leader,
@@ -69,6 +67,11 @@ pub struct LogResponse {
     pub ok: bool,
 }
 
+#[derive(SerialDecodable, SerialEncodable, Clone, Debug)]
+pub struct NodeIdMsg {
+    pub id: NodeId,
+}
+
 impl VoteResponse {
     pub fn set_ok(&mut self, ok: bool) {
         self.ok = ok;
@@ -85,15 +88,7 @@ pub struct Log {
 }
 
 #[derive(Clone, Debug, Eq, PartialEq, Hash, SerialDecodable, SerialEncodable)]
-pub struct NodeId(pub Vec<u8>);
-
-impl From<Url> for NodeId {
-    fn from(addr: Url) -> Self {
-        let ser = serialize(&addr);
-        let hash = blake3::hash(&ser).as_bytes().to_vec();
-        Self(hash)
-    }
-}
+pub struct NodeId(pub String);
 
 #[derive(Clone, Debug, SerialDecodable, SerialEncodable)]
 pub struct Logs(pub Vec<Log>);
@@ -166,9 +161,7 @@ pub enum NetMsgMethod {
     VoteResponse = 2,
     VoteRequest = 3,
     BroadcastRequest = 4,
-    // this only used for listener node
-    SyncRequest = 5,
-    SyncResponse = 6,
+    NodeIdMsg = 5,
 }
 
 impl Encodable for NetMsgMethod {
@@ -179,8 +172,7 @@ impl Encodable for NetMsgMethod {
             Self::VoteResponse => 2,
             Self::VoteRequest => 3,
             Self::BroadcastRequest => 4,
-            Self::SyncRequest => 5,
-            Self::SyncResponse => 6,
+            Self::NodeIdMsg => 5,
         };
         (len as u8).encode(s)
     }
@@ -195,8 +187,7 @@ impl Decodable for NetMsgMethod {
             2 => Self::VoteResponse,
             3 => Self::VoteRequest,
             4 => Self::BroadcastRequest,
-            5 => Self::SyncRequest,
-            _ => Self::SyncResponse,
+            _ => Self::NodeIdMsg,
         })
     }
 }

src/raft/protocol_raft.rs

@@ -2,25 +2,28 @@ use async_std::sync::{Arc, Mutex};
 
 use async_executor::Executor;
 use async_trait::async_trait;
+use chrono::Utc;
 use fxhash::FxHashMap;
 use log::debug;
+use rand::{rngs::OsRng, RngCore};
 
-use crate::{net, Result};
+use crate::{net, util::serial::serialize, Result};
 
-use super::primitives::{NetMsg, NodeId};
+use super::primitives::{NetMsg, NetMsgMethod, NodeId, NodeIdMsg};
 
 pub struct ProtocolRaft {
-    id: Option<NodeId>,
+    id: NodeId,
     jobsman: net::ProtocolJobsManagerPtr,
     notify_queue_sender: async_channel::Sender<NetMsg>,
     msg_sub: net::MessageSubscription<NetMsg>,
     p2p: net::P2pPtr,
     seen_msgs: Arc<Mutex<FxHashMap<String, i64>>>,
+    channel: net::ChannelPtr,
 }
 
 impl ProtocolRaft {
     pub async fn init(
-        id: Option<NodeId>,
+        id: NodeId,
         channel: net::ChannelPtr,
         notify_queue_sender: async_channel::Sender<NetMsg>,
         p2p: net::P2pPtr,
@@ -35,21 +38,37 @@ impl ProtocolRaft {
             id,
             notify_queue_sender,
             msg_sub,
-            jobsman: net::ProtocolJobsManager::new("ProtocolRaft", channel),
+            jobsman: net::ProtocolJobsManager::new("ProtocolRaft", channel.clone()),
             p2p,
             seen_msgs,
+            channel,
         })
     }
 
     async fn handle_receive_msg(self: Arc<Self>) -> Result<()> {
-        debug!(target: "raft", "ProtocolRaft::handle_receive_msg() [START]");
+        debug!(target: "protocol_raft", "ProtocolRaft::handle_receive_msg() [START]");
+
+        // on initialization send a NodeIdMsg
+        let random_id = OsRng.next_u64();
+        let node_id_msg = serialize(&NodeIdMsg { id: self.id.clone() });
+        let net_msg = NetMsg {
+            id: random_id,
+            recipient_id: None,
+            payload: node_id_msg.to_vec(),
+            method: NetMsgMethod::NodeIdMsg,
+        };
+        {
+            self.seen_msgs.lock().await.insert(random_id.to_string(), Utc::now().timestamp());
+        }
+        self.channel.send(net_msg).await?;
+
         loop {
             let msg = self.msg_sub.receive().await?;
 
             debug!(
-                target: "raft",
-                "ProtocolRaft::handle_receive_msg() received id: {:?} method {:?}",
-                &msg.id, &msg.method
+            target: "protocol_raft",
+            "ProtocolRaft::handle_receive_msg() received id: {:?} method {:?}",
+            &msg.id, &msg.method
             );
 
             {
@@ -63,13 +82,10 @@ impl ProtocolRaft {
             let msg = (*msg).clone();
             self.p2p.broadcast(msg.clone()).await?;
 
-            // check if the ids are equal when both
-            // the local node and recipient ids are Some(id)
-            if let Some(self_id) = &self.id {
-                if let Some(recipient_id) = &msg.recipient_id {
-                    if self_id != recipient_id {
-                        continue
-                    }
+            // check if the local node and recipient id are equal
+            if let Some(recipient_id) = &msg.recipient_id {
+                if &self.id != recipient_id {
+                    continue
                 }
             }
 
@@ -84,10 +100,10 @@ impl net::ProtocolBase for ProtocolRaft {
     /// protocol task manager, then queues the reply. Sends out a ping and
     /// waits for pong reply. Waits for ping and replies with a pong.
     async fn start(self: Arc<Self>, executor: Arc<Executor<'_>>) -> Result<()> {
-        debug!(target: "raft", "ProtocolRaft::start() [START]");
+        debug!(target: "protocol_raft", "ProtocolRaft::start() [START]");
         self.jobsman.clone().start(executor.clone());
         self.jobsman.clone().spawn(self.clone().handle_receive_msg(), executor.clone()).await;
-        debug!(target: "raft", "ProtocolRaft::start() [END]");
+        debug!(target: "protocol_raft", "ProtocolRaft::start() [END]");
         Ok(())
     }
 

src/raft/settings.rs

@@ -0,0 +1,36 @@
+use std::path::PathBuf;
+
+#[derive(Clone, Debug)]
+pub struct RaftSettings {
+    //
+    // Milliseconds
+    //
+    pub heartbeat_timeout: u64,
+    pub timeout: u64,
+
+    //
+    // Seconds
+    //
+    pub prun_messages_duration: i64,
+    pub prun_nodes_ids_duration: i64,
+    // must be greater than (timeout * 2)
+    pub node_id_timeout: i64,
+
+    //
+    // Datastore path
+    //
+    pub datastore_path: PathBuf,
+}
+
+impl Default for RaftSettings {
+    fn default() -> Self {
+        Self {
+            heartbeat_timeout: 500,
+            timeout: 6000,
+            prun_messages_duration: 120,
+            prun_nodes_ids_duration: 120,
+            node_id_timeout: 16,
+            datastore_path: PathBuf::from(""),
+        }
+    }
+}