raft: Remove entire raft module.

This code is not needed anymore.

script/research/raft-diag/.gitignore

@@ -1,2 +0,0 @@
-/target
-Cargo.lock

script/research/raft-diag/Cargo.toml

@@ -1,37 +0,0 @@
-[package]
-name = "raft-diag"
-version = "0.4.1"
-authors = ["Dyne.org foundation <foundation@dyne.org>"]
-edition = "2021"
-
-[workspace]
-
-[dependencies]
-darkfi = {path = "../../../", features = ["raft"]}
-
-# Async
-smol = "1.3.0"
-async-std = {version = "1.12.0", features = ["attributes"]}
-async-trait = "0.1.68"
-async-channel = "1.8.0"
-async-executor = "1.5.1"
-easy-parallel = "3.3.0"
-futures = "0.3.28"
-
-# Misc
-log = "0.4.19"
-simplelog = "0.12.1"
-rand = "0.8.5"
-chrono = "0.4.26"
-thiserror = "1.0.40"
-ctrlc = { version = "3.4.0", features = ["termination"] }
-url = "2.4.0"
-fxhash = "0.2.1"
-
-# Encoding and parsing
-serde = {version = "1.0.164", features = ["derive"]}
-serde_json = "1.0.96"
-structopt = "0.3.26"
-hex = "0.4.3"
-bs58 = "0.5.0"
-toml = "0.7.4"

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

@@ -1,246 +0,0 @@
-/* This file is part of DarkFi (https://dark.fi)
- *
- * Copyright (C) 2020-2023 Dyne.org foundation
- *
- * This program is free software: you can redistribute it and/or modify
- * it under the terms of the GNU Affero General Public License as
- * published by the Free Software Foundation, either version 3 of the
- * License, or (at your option) any later version.
- *
- * This program is distributed in the hope that it will be useful,
- * but WITHOUT ANY WARRANTY; without even the implied warranty of
- * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
- * GNU Affero General Public License for more details.
- *
- * You should have received a copy of the GNU Affero General Public License
- * along with this program.  If not, see <https://www.gnu.org/licenses/>.
- */
-
-use async_std::sync::{Arc, Mutex};
-use std::path::Path;
-
-use async_executor::Executor;
-use fxhash::FxHashMap;
-use log::{error, info, warn};
-use smol::future;
-use structopt::StructOpt;
-use url::Url;
-
-use darkfi::{
-    net,
-    raft::{DataStore, NetMsg, ProtocolRaft, Raft, RaftSettings},
-    util::{
-        cli::{get_log_config, get_log_level},
-        expand_path,
-        serial::{SerialDecodable, SerialEncodable},
-        sleep,
-    },
-    Result,
-};
-
-#[derive(Clone, Debug, StructOpt)]
-#[structopt(name = "raft-diag")]
-pub struct Args {
-    /// JSON-RPC listen URL
-    #[structopt(long = "rpc", default_value = "tcp://127.0.0.1:12055")]
-    pub rpc_listen: Url,
-    /// Inbound listen URL
-    #[structopt(long = "inbound")]
-    pub inbound_url: Vec<Url>,
-    /// Seed Urls
-    #[structopt(long = "seeds")]
-    pub seed_urls: Vec<Url>,
-    /// Outbound connections
-    #[structopt(long = "outbound", default_value = "0")]
-    pub outbound_connections: u32,
-    /// Sets Datastore Path
-    #[structopt(long = "path", default_value = "test1.db")]
-    pub datastore: String,
-    /// Check if all datastore paths provided are synced
-    #[structopt(long = "check")]
-    pub check: Vec<String>,
-    /// Datastore path to extract and print it
-    #[structopt(long = "extract")]
-    pub extract: Option<String>,
-    /// Number of messages to broadcast
-    #[structopt(short, default_value = "0")]
-    pub broadcast: u32,
-    /// Increase verbosity
-    #[structopt(short, parse(from_occurrences))]
-    pub verbose: u8,
-}
-
-#[derive(Debug, Clone, SerialEncodable, SerialDecodable, PartialEq, Eq)]
-pub struct Message {
-    payload: String,
-}
-
-fn extract(path: &str) -> Result<()> {
-    if !Path::new(path).exists() {
-        return Ok(())
-    }
-
-    let db = DataStore::<Message>::new(path)?;
-    let commits = db.commits.get_all()?;
-
-    println!("{:?}", commits);
-
-    Ok(())
-}
-
-fn check(args: Args) -> Result<()> {
-    let mut commits_check = vec![];
-
-    for path in args.check {
-        if !Path::new(&path).exists() {
-            continue
-        }
-        let db = DataStore::<Message>::new(&path)?;
-        let commits = db.commits.get_all()?;
-        commits_check.push(commits);
-    }
-
-    let result = commits_check.windows(2).all(|w| w[0] == w[1]);
-
-    println!("Synced: {}", result);
-
-    Ok(())
-}
-
-async fn start_broadcasting(n: u32, sender: async_channel::Sender<Message>) -> Result<()> {
-    sleep(8).await;
-    info!(target: "raft", "Start broadcasting...");
-    for id in 0..n {
-        let msg = format!("msg_test_{}", id);
-        info!(target: "raft", "Send a message {:?}", msg);
-        let msg = Message { payload: msg };
-        sender.send(msg).await?;
-    }
-
-    Ok(())
-}
-
-async fn receive_loop(receiver: async_channel::Receiver<Message>) -> Result<()> {
-    loop {
-        let msg = receiver.recv().await?;
-        info!(target: "raft", "Receive new msg {:?}", msg);
-    }
-}
-
-async fn start(args: Args, executor: Arc<Executor<'_>>) -> Result<()> {
-    let net_settings = net::Settings {
-        outbound_connections: args.outbound_connections,
-        inbound: args.inbound_url.clone(),
-        external_addr: args.inbound_url,
-        seeds: args.seed_urls,
-        ..net::Settings::default()
-    };
-
-    //
-    // Raft
-    //
-
-    let datastore_raft = expand_path(&args.datastore)?;
-
-    let seen_net_msgs = Arc::new(Mutex::new(FxHashMap::default()));
-
-    let raft_settings = RaftSettings { datastore_path: datastore_raft, ..RaftSettings::default() };
-
-    let mut raft = Raft::<Message>::new(raft_settings, seen_net_msgs.clone())?;
-
-    //
-    // P2p setup
-    //
-
-    let (p2p_send_channel, p2p_recv_channel) = async_channel::unbounded::<NetMsg>();
-
-    let p2p = net::P2p::new(net_settings).await;
-    let p2p = p2p.clone();
-
-    let registry = p2p.protocol_registry();
-
-    let raft_node_id = raft.id();
-    registry
-        .register(net::SESSION_ALL, move |channel, p2p| {
-            let raft_node_id = raft_node_id.clone();
-            let sender = p2p_send_channel.clone();
-            let seen_net_msgs_cloned = seen_net_msgs.clone();
-            async move {
-                ProtocolRaft::init(raft_node_id, channel, sender, p2p, seen_net_msgs_cloned).await
-            }
-        })
-        .await;
-
-    p2p.clone().start(executor.clone()).await?;
-
-    executor.spawn(p2p.clone().run(executor.clone())).detach();
-
-    //
-    // Waiting Exit signal
-    //
-    let (signal, shutdown) = async_channel::bounded::<()>(1);
-    ctrlc::set_handler(move || {
-        warn!("Catch exit signal");
-        // cleaning up tasks running in the background
-        if let Err(e) = async_std::task::block_on(signal.send(())) {
-            error!("Error on sending exit signal: {}", e);
-        }
-    })
-    .unwrap();
-
-    if args.broadcast != 0 {
-        executor.spawn(start_broadcasting(args.broadcast, raft.sender())).detach();
-    }
-
-    executor.spawn(receive_loop(raft.receiver())).detach();
-
-    raft.run(p2p.clone(), p2p_recv_channel.clone(), executor.clone(), shutdown.clone()).await?;
-    Ok(())
-}
-
-fn main() -> Result<()> {
-    let args = Args::from_args();
-    let log_level = get_log_level(args.verbose.into());
-    let log_config = get_log_config();
-
-    let mut log_path = expand_path(&args.datastore)?;
-    let log_name: String = log_path.file_name().as_ref().unwrap().to_str().unwrap().to_owned();
-    log_path.pop();
-    let log_path = log_path.join(&format!("{}.log", log_name));
-    let env_log_file_path = std::fs::File::create(log_path).unwrap();
-
-    simplelog::CombinedLogger::init(vec![
-        simplelog::TermLogger::new(
-            log_level,
-            log_config.clone(),
-            simplelog::TerminalMode::Mixed,
-            simplelog::ColorChoice::Auto,
-        ),
-        simplelog::WriteLogger::new(log_level, log_config, env_log_file_path),
-    ])?;
-
-    if !args.check.is_empty() {
-        return check(args)
-    }
-
-    if args.extract.is_some() {
-        return extract(&args.extract.unwrap())
-    }
-
-    // https://docs.rs/smol/latest/smol/struct.Executor.html#examples
-    let ex = Arc::new(async_executor::Executor::new());
-    let (signal, shutdown) = async_channel::unbounded::<()>();
-    let (_, result) = easy_parallel::Parallel::new()
-        // Run four executor threads
-        .each(0..4, |_| future::block_on(ex.run(shutdown.recv())))
-        // Run the main future on the current thread.
-        .finish(|| {
-            future::block_on(async {
-                start(args, ex.clone()).await?;
-                drop(signal);
-                Ok::<(), darkfi::Error>(())
-            })
-        });
-
-    result
-}

script/research/raft-diag/tmux_sessions.sh

@@ -1,12 +0,0 @@
-#!/bin/sh
-export LOG_TARGETS='raft' 
-
-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 -v "
-sleep 2
-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 -v"
-tmux attach

src/lib.rs

@@ -28,8 +28,6 @@ pub mod consensus;
 #[cfg(feature = "blockchain")]
 pub mod validator;
 
-#[cfg(feature = "dht")]
-pub mod dht;
 #[cfg(feature = "dht")]
 pub mod dht2;
 
@@ -39,9 +37,6 @@ pub mod event_graph;
 #[cfg(feature = "net")]
 pub mod net;
 
-#[cfg(feature = "raft")]
-pub mod raft;
-
 #[cfg(feature = "rpc")]
 pub mod rpc;
 

src/raft/consensus.rs

@@ -1,388 +0,0 @@
-/* This file is part of DarkFi (https://dark.fi)
- *
- * Copyright (C) 2020-2023 Dyne.org foundation
- *
- * This program is free software: you can redistribute it and/or modify
- * it under the terms of the GNU Affero General Public License as
- * published by the Free Software Foundation, either version 3 of the
- * License, or (at your option) any later version.
- *
- * This program is distributed in the hope that it will be useful,
- * but WITHOUT ANY WARRANTY; without even the implied warranty of
- * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
- * GNU Affero General Public License for more details.
- *
- * You should have received a copy of the GNU Affero General Public License
- * along with this program.  If not, see <https://www.gnu.org/licenses/>.
- */
-
-use std::{collections::HashMap, time::Duration};
-
-use async_std::{
-    sync::{Arc, Mutex},
-    task::sleep,
-};
-use chrono::Utc;
-use darkfi_serial::{deserialize, serialize, Decodable, Encodable};
-use futures::{select, FutureExt};
-use log::{debug, error, warn};
-use rand::{distributions::Alphanumeric, rngs::OsRng, thread_rng, Rng, RngCore};
-use smol::Executor;
-
-use crate::{net, Error, Result};
-
-use super::{
-    p2p_send_loop,
-    primitives::{
-        BroadcastMsgRequest, Channel, Log, LogRequest, LogResponse, Logs, MapLength, NetMsg,
-        NetMsgMethod, NodeId, NodeIdMsg, Role, Sender, VoteRequest, VoteResponse,
-    },
-    prune_map, DataStore, RaftSettings,
-};
-
-async fn send_loop(sender: smol::channel::Sender<()>, timeout: Duration) -> Result<()> {
-    loop {
-        sleep(timeout).await;
-        sender.send(()).await?;
-    }
-}
-
-pub fn gen_id(len: usize) -> String {
-    thread_rng().sample_iter(&Alphanumeric).take(len).map(char::from).collect()
-}
-
-pub struct Raft<T> {
-    id: NodeId,
-
-    pub(super) role: Role,
-
-    pub(super) current_leader: NodeId,
-
-    pub(super) votes_received: Vec<NodeId>,
-
-    pub(super) sent_length: MapLength,
-    pub(super) acked_length: MapLength,
-
-    pub(super) nodes: Arc<Mutex<HashMap<NodeId, i64>>>,
-
-    pub(super) last_term: u64,
-
-    pub(super) last_heartbeat: i64,
-
-    p2p_sender: Sender,
-
-    msgs_channel: Channel<T>,
-    commits_channel: Channel<T>,
-
-    datastore: DataStore<T>,
-
-    seen_msgs: Arc<Mutex<HashMap<String, i64>>>,
-
-    pub(super) settings: RaftSettings,
-
-    pending_msgs: Vec<T>,
-}
-
-impl<T: Decodable + Encodable + Clone> Raft<T> {
-    pub fn new(
-        settings: RaftSettings,
-        seen_msgs: Arc<Mutex<HashMap<String, i64>>>,
-    ) -> Result<Self> {
-        if settings.datastore_path.to_str().is_none() {
-            error!(target: "raft::consensus", "datastore path is incorrect");
-            return Err(Error::ParseFailed("unable to parse pathbuf to str"))
-        };
-
-        let datastore = DataStore::new(settings.datastore_path.to_str().unwrap())?;
-
-        // broadcasting channels
-        let msgs_channel = smol::channel::unbounded::<T>();
-        let commits_channel = smol::channel::unbounded::<T>();
-
-        let p2p_sender = smol::channel::unbounded::<NetMsg>();
-
-        let id = match datastore.id.get_last()? {
-            Some(_id) => _id,
-            None => {
-                // FIXME: This should be a big number, like a hash.
-                let id = NodeId(gen_id(30));
-                datastore.id.insert(&id)?;
-                id
-            }
-        };
-
-        let role = Role::Follower;
-
-        Ok(Self {
-            id,
-            role,
-            current_leader: NodeId("".into()),
-            votes_received: vec![],
-            sent_length: MapLength(HashMap::default()),
-            acked_length: MapLength(HashMap::default()),
-            nodes: Arc::new(Mutex::new(HashMap::default())),
-            last_term: 0,
-            last_heartbeat: Utc::now().timestamp(),
-            p2p_sender,
-            msgs_channel,
-            commits_channel,
-            datastore,
-            seen_msgs,
-            settings,
-            pending_msgs: vec![],
-        })
-    }
-
-    ///  
-    ///  Run raft consensus and wait stop_signal channel to terminate
-    ///
-    pub async fn run(
-        &mut self,
-        p2p: net::P2pPtr,
-        p2p_recv_channel: smol::channel::Receiver<NetMsg>,
-        executor: Arc<Executor<'_>>,
-        stop_signal: smol::channel::Receiver<()>,
-    ) -> Result<()> {
-        let p2p_send_task = executor.spawn(p2p_send_loop(self.p2p_sender.1.clone(), p2p.clone()));
-
-        let prune_seen_messages_task = executor
-            .spawn(prune_map::<String>(self.seen_msgs.clone(), self.settings.prun_duration));
-
-        let prune_nodes_id_task =
-            executor.spawn(prune_map::<NodeId>(self.nodes.clone(), self.settings.prun_duration));
-
-        let (id_sx, id_rv) = smol::channel::unbounded::<()>();
-        let (heartbeat_sx, heartbeat_rv) = smol::channel::unbounded::<()>();
-        let (timeout_sx, timeout_rv) = smol::channel::unbounded::<()>();
-
-        let id_timeout = Duration::from_secs(self.settings.id_timeout);
-        let send_id_task = executor.spawn(send_loop(id_sx, id_timeout));
-
-        let heartbeat_timeout = Duration::from_millis(self.settings.heartbeat_timeout);
-        let send_heartbeat_task = executor.spawn(send_loop(heartbeat_sx, heartbeat_timeout));
-
-        let rng = &mut OsRng;
-        let timeout =
-            Duration::from_secs(rng.gen_range(0..self.settings.timeout) + self.settings.timeout);
-        let send_timeout_task = executor.spawn(send_loop(timeout_sx, timeout));
-
-        let broadcast_msg_rv = self.msgs_channel.1.clone();
-
-        loop {
-            let mut result = select! {
-                m =  p2p_recv_channel.recv().fuse() => self.handle_method(m?).await,
-                m =  broadcast_msg_rv.recv().fuse() => self.broadcast_msg(&m?,None).await,
-                _ =  id_rv.recv().fuse() => self.send_id_msg().await,
-                _ = heartbeat_rv.recv().fuse() => self.send_heartbeat().await,
-                _ = timeout_rv.recv().fuse() => self.send_vote_request().await,
-                _ = stop_signal.recv().fuse() => break,
-            };
-
-            // send pending messages
-            if !self.pending_msgs.is_empty() && self.role != Role::Candidate {
-                let pending_msgs = self.pending_msgs.clone();
-                for m in &pending_msgs {
-                    result = self.broadcast_msg(m, None).await;
-                }
-                self.pending_msgs = vec![];
-            }
-
-            if let Err(e) = result {
-                warn!(target: "raft::consensus", "warn: {}", e);
-            }
-        }
-
-        warn!(target: "raft::consensus", "Raft Terminating...");
-        p2p_send_task.cancel().await;
-        prune_seen_messages_task.cancel().await;
-        prune_nodes_id_task.cancel().await;
-        send_id_task.cancel().await;
-        send_heartbeat_task.cancel().await;
-        send_timeout_task.cancel().await;
-        self.datastore.flush().await?;
-        Ok(())
-    }
-
-    ///  
-    /// Return async receiver channel which can be used to receive T Messages
-    /// from raft consensus
-    ///
-    pub fn receiver(&self) -> smol::channel::Receiver<T> {
-        self.commits_channel.1.clone()
-    }
-
-    ///  
-    /// Return async sender channel which can be used to broadcast T Messages
-    /// to raft consensus
-    ///
-    pub fn sender(&self) -> smol::channel::Sender<T> {
-        self.msgs_channel.0.clone()
-    }
-
-    ///  
-    /// Return the raft node id
-    ///
-    pub fn id(&self) -> NodeId {
-        self.id.clone()
-    }
-
-    async fn send_id_msg(&self) -> Result<()> {
-        let id_msg = serialize(&NodeIdMsg { id: self.id.clone() });
-        self.send(None, &id_msg, NetMsgMethod::NodeIdMsg, None).await?;
-        Ok(())
-    }
-
-    async fn broadcast_msg(&mut self, msg: &T, msg_id: Option<u64>) -> Result<()> {
-        match self.role {
-            Role::Leader => {
-                let msg = serialize(msg);
-                let log = Log { msg, term: self.current_term()? };
-                self.push_log(&log)?;
-                self.acked_length.insert(&self.id, self.logs_len());
-            }
-            Role::Follower => {
-                let b_msg = BroadcastMsgRequest(serialize(msg));
-                self.send(
-                    Some(self.current_leader.clone()),
-                    &serialize(&b_msg),
-                    NetMsgMethod::BroadcastRequest,
-                    msg_id,
-                )
-                .await?;
-            }
-            Role::Candidate => {
-                self.pending_msgs.push(msg.clone());
-            }
-        }
-
-        debug!(target: "raft::consensus", "Role: {:?} Id: {:?}, broadcast a msg id: {:?} ", self.role, self.id, msg_id);
-
-        Ok(())
-    }
-
-    async fn handle_method(&mut self, msg: NetMsg) -> Result<()> {
-        match msg.method {
-            NetMsgMethod::LogResponse => {
-                let lr: LogResponse = deserialize(&msg.payload)?;
-                self.receive_log_response(lr).await?;
-            }
-            NetMsgMethod::LogRequest => {
-                self.last_heartbeat = Utc::now().timestamp();
-                let lr: LogRequest = deserialize(&msg.payload)?;
-                self.receive_log_request(lr).await?;
-            }
-            NetMsgMethod::VoteResponse => {
-                let vr: VoteResponse = deserialize(&msg.payload)?;
-                self.receive_vote_response(vr).await?;
-            }
-            NetMsgMethod::VoteRequest => {
-                let vr: VoteRequest = deserialize(&msg.payload)?;
-                self.receive_vote_request(vr).await?;
-            }
-            NetMsgMethod::BroadcastRequest => {
-                let vr: BroadcastMsgRequest = deserialize(&msg.payload)?;
-                let d: T = deserialize(&vr.0)?;
-                self.broadcast_msg(&d, Some(msg.id)).await?;
-            }
-            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());
-                }
-            }
-        }
-
-        debug!(target: "raft::consensus", "Role: {:?} Id: {:?}, receive a msg with id: {}  recipient_id: {:?} method: {:?} ",
-               self.role, self.id, msg.id, &msg.recipient_id, &msg.method);
-        Ok(())
-    }
-
-    pub(super) async fn send(
-        &self,
-        recipient_id: Option<NodeId>,
-        payload: &[u8],
-        method: NetMsgMethod,
-        msg_id: Option<u64>,
-    ) -> Result<()> {
-        let random_id = if msg_id.is_some() { msg_id.unwrap() } else { OsRng.next_u64() };
-
-        debug!(target: "raft::consensus","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());
-        self.p2p_sender.0.send(net_msg).await?;
-
-        Ok(())
-    }
-
-    pub(super) fn reset_last_term(&mut self) -> Result<()> {
-        self.last_term = 0;
-
-        if let Some(log) = self.last_log()? {
-            self.last_term = log.term;
-        }
-
-        Ok(())
-    }
-
-    pub(super) fn set_current_term(&mut self, i: &u64) -> Result<()> {
-        self.datastore.current_term.insert(i)
-    }
-
-    pub(super) fn set_voted_for(&mut self, i: &Option<NodeId>) -> Result<()> {
-        self.datastore.voted_for.insert(i)
-    }
-
-    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)
-    }
-
-    pub(super) fn push_log(&mut self, log: &Log) -> Result<()> {
-        self.datastore.logs.insert(log)
-    }
-
-    pub(super) fn push_logs(&mut self, logs: &Logs) -> Result<()> {
-        self.datastore.logs.wipe_insert_all(&logs.to_vec())
-    }
-
-    pub(super) fn current_term(&self) -> Result<u64> {
-        Ok(self.datastore.current_term.get_last()?.unwrap_or(0))
-    }
-
-    pub(super) fn voted_for(&self) -> Result<Option<NodeId>> {
-        Ok(self.datastore.voted_for.get_last()?.flatten())
-    }
-
-    pub(super) fn commits_len(&self) -> u64 {
-        self.datastore.commits.len()
-    }
-
-    fn logs(&self) -> Result<Logs> {
-        Ok(Logs(self.datastore.logs.get_all()?))
-    }
-
-    pub(super) fn logs_len(&self) -> u64 {
-        self.datastore.logs.len()
-    }
-
-    fn last_log(&self) -> Result<Option<Log>> {
-        self.datastore.logs.get_last()
-    }
-
-    pub(super) fn get_log(&self, index: u64) -> Result<Log> {
-        self.datastore.logs.get(index)
-    }
-
-    pub(super) fn slice_logs_from(&self, index: u64) -> Result<Option<Logs>> {
-        let logs = self.logs()?;
-        Ok(logs.slice_from(index))
-    }
-
-    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

@@ -1,94 +0,0 @@
-/* This file is part of DarkFi (https://dark.fi)
- *
- * Copyright (C) 2020-2023 Dyne.org foundation
- *
- * This program is free software: you can redistribute it and/or modify
- * it under the terms of the GNU Affero General Public License as
- * published by the Free Software Foundation, either version 3 of the
- * License, or (at your option) any later version.
- *
- * This program is distributed in the hope that it will be useful,
- * but WITHOUT ANY WARRANTY; without even the implied warranty of
- * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
- * GNU Affero General Public License for more details.
- *
- * You should have received a copy of the GNU Affero General Public License
- * along with this program.  If not, see <https://www.gnu.org/licenses/>.
- */
-
-use chrono::Utc;
-use darkfi_serial::{serialize, Decodable, Encodable};
-use log::info;
-
-use crate::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<()> {
-        if self.role == Role::Leader {
-            return Ok(())
-        }
-
-        let last_heartbeat_duration = Utc::now().timestamp() - self.last_heartbeat;
-
-        if last_heartbeat_duration < self.settings.timeout as i64 {
-            return Ok(())
-        }
-
-        self.set_current_term(&(self.current_term()? + 1))?;
-
-        if self.role != Role::Candidate {
-            info!(target: "raft::consensus_candidate", "Set the node role as Candidate");
-            self.role = Role::Candidate;
-        }
-
-        self.set_voted_for(&Some(self.id()))?;
-        self.votes_received = vec![self.id()];
-
-        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 {
-            if self.votes_received.contains(&vr.node_id) {
-                return 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) {
-                info!(target: "raft::consensus_candidate", "Set the node role as Leader");
-                self.role = Role::Leader;
-                self.current_leader = self.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

@@ -1,127 +0,0 @@
-/* This file is part of DarkFi (https://dark.fi)
- *
- * Copyright (C) 2020-2023 Dyne.org foundation
- *
- * This program is free software: you can redistribute it and/or modify
- * it under the terms of the GNU Affero General Public License as
- * published by the Free Software Foundation, either version 3 of the
- * License, or (at your option) any later version.
- *
- * This program is distributed in the hope that it will be useful,
- * but WITHOUT ANY WARRANTY; without even the implied warranty of
- * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
- * GNU Affero General Public License for more details.
- *
- * You should have received a copy of the GNU Affero General Public License
- * along with this program.  If not, see <https://www.gnu.org/licenses/>.
- */
-
-use std::cmp::min;
-
-use darkfi_serial::{serialize, Decodable, Encodable};
-use log::debug;
-
-use super::{
-    primitives::{LogRequest, LogResponse, Logs, NetMsgMethod, Role, VoteRequest, VoteResponse},
-    Raft,
-};
-use crate::Result;
-
-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.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::consensus_follower",
-        "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.id(), current_term: self.current_term()?, ack, ok };
-
-        debug!(target: "raft::consensus_follower",
-         "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

@@ -1,136 +0,0 @@
-/* This file is part of DarkFi (https://dark.fi)
- *
- * Copyright (C) 2020-2023 Dyne.org foundation
- *
- * This program is free software: you can redistribute it and/or modify
- * it under the terms of the GNU Affero General Public License as
- * published by the Free Software Foundation, either version 3 of the
- * License, or (at your option) any later version.
- *
- * This program is distributed in the hope that it will be useful,
- * but WITHOUT ANY WARRANTY; without even the implied warranty of
- * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
- * GNU Affero General Public License for more details.
- *
- * You should have received a copy of the GNU Affero General Public License
- * along with this program.  If not, see <https://www.gnu.org/licenses/>.
- */
-use std::collections::HashMap;
-
-use darkfi_serial::{serialize, Decodable, Encodable};
-
-use crate::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<()> {
-        if self.role != Role::Leader {
-            return Ok(())
-        }
-
-        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.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: HashMap<NodeId, i64>, length: u64) -> HashMap<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;
-        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

@@ -1,140 +0,0 @@
-/* This file is part of DarkFi (https://dark.fi)
- *
- * Copyright (C) 2020-2023 Dyne.org foundation
- *
- * This program is free software: you can redistribute it and/or modify
- * it under the terms of the GNU Affero General Public License as
- * published by the Free Software Foundation, either version 3 of the
- * License, or (at your option) any later version.
- *
- * This program is distributed in the hope that it will be useful,
- * but WITHOUT ANY WARRANTY; without even the implied warranty of
- * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
- * GNU Affero General Public License for more details.
- *
- * You should have received a copy of the GNU Affero General Public License
- * along with this program.  If not, see <https://www.gnu.org/licenses/>.
- */
-
-use std::marker::PhantomData;
-
-use darkfi_serial::{deserialize, serialize, Decodable, Encodable};
-use log::debug;
-use sled::Batch;
-
-use crate::{Error, Result};
-
-use super::primitives::{Log, NodeId};
-
-const SLED_LOGS_TREE: &[u8] = b"_logs";
-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,
-    pub logs: DataTree<Log>,
-    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> {
-    pub fn new(db_path: &str) -> Result<Self> {
-        let _db = sled::open(db_path)?;
-        let logs = DataTree::new(&_db, SLED_LOGS_TREE)?;
-        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, id })
-    }
-    pub async fn flush(&self) -> Result<()> {
-        debug!(target: "raft::datastore", "DataStore flush");
-        self._db.flush_async().await?;
-        Ok(())
-    }
-}
-
-pub struct DataTree<T> {
-    tree: sled::Tree,
-    phantom: PhantomData<T>,
-}
-
-impl<T: Decodable + Encodable> DataTree<T> {
-    pub fn new(db: &sled::Db, tree_name: &[u8]) -> Result<Self> {
-        let tree = db.open_tree(tree_name)?;
-        Ok(Self { tree, phantom: PhantomData })
-    }
-
-    pub fn insert(&self, data: &T) -> Result<()> {
-        let serialized = serialize(data);
-        let last_index: u64 = if let Some(d) = self.tree.last()? {
-            u64::from_be_bytes(d.0.to_vec().try_into().unwrap()) + 1
-        } else {
-            0
-        };
-
-        self.tree.insert(last_index.to_be_bytes(), serialized)?;
-        Ok(())
-    }
-
-    pub fn wipe_insert_all(&self, data: &[T]) -> Result<()> {
-        self.tree.clear()?;
-
-        let mut batch = Batch::default();
-
-        for (i, d) in data.iter().enumerate() {
-            let serialized = serialize(d);
-            batch.insert(&(i as u64).to_be_bytes(), serialized);
-        }
-
-        self.tree.apply_batch(batch)?;
-
-        Ok(())
-    }
-
-    pub fn get_all(&self) -> Result<Vec<T>> {
-        let mut ret: Vec<T> = Vec::new();
-
-        for i in self.tree.iter() {
-            let da = deserialize(&i?.1)?;
-            ret.push(da)
-        }
-
-        Ok(ret)
-    }
-
-    pub fn len(&self) -> u64 {
-        self.tree.len() as u64
-    }
-
-    pub fn get_last(&self) -> Result<Option<T>> {
-        if let Some(found) = self.tree.last()? {
-            let da = deserialize(&found.1)?;
-            return Ok(Some(da))
-        }
-        Ok(None)
-    }
-
-    pub fn get(&self, index: u64) -> Result<T> {
-        let index_bytes = index.to_be_bytes();
-        if let Some(found) = self.tree.get(index_bytes)? {
-            let da = deserialize(&found)?;
-            return Ok(da)
-        }
-        Err(Error::RaftError(format!(
-            "Unable to get the item with index {} {:?}",
-            index,
-            self.is_empty()
-        )))
-    }
-
-    pub fn is_empty(&self) -> bool {
-        self.tree.is_empty()
-    }
-}

src/raft/mod.rs

@@ -1,70 +0,0 @@
-/* This file is part of DarkFi (https://dark.fi)
- *
- * Copyright (C) 2020-2023 Dyne.org foundation
- *
- * This program is free software: you can redistribute it and/or modify
- * it under the terms of the GNU Affero General Public License as
- * published by the Free Software Foundation, either version 3 of the
- * License, or (at your option) any later version.
- *
- * This program is distributed in the hope that it will be useful,
- * but WITHOUT ANY WARRANTY; without even the implied warranty of
- * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
- * GNU Affero General Public License for more details.
- *
- * You should have received a copy of the GNU Affero General Public License
- * along with this program.  If not, see <https://www.gnu.org/licenses/>.
- */
-
-use std::collections::HashMap;
-
-use async_std::sync::{Arc, Mutex};
-use chrono::Utc;
-use log::{debug, error};
-
-use crate::{net, util::async_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::{gen_id, Raft};
-pub use datastore::DataStore;
-pub use primitives::NetMsg;
-pub use protocol_raft::ProtocolRaft;
-pub use settings::RaftSettings;
-
-// 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<HashMap<T, i64>>>,
-    seen_duration: i64,
-) {
-    loop {
-        async_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: smol::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
-        }
-    }
-}

src/raft/primitives.rs

@@ -1,207 +0,0 @@
-/* This file is part of DarkFi (https://dark.fi)
- *
- * Copyright (C) 2020-2023 Dyne.org foundation
- *
- * This program is free software: you can redistribute it and/or modify
- * it under the terms of the GNU Affero General Public License as
- * published by the Free Software Foundation, either version 3 of the
- * License, or (at your option) any later version.
- *
- * This program is distributed in the hope that it will be useful,
- * but WITHOUT ANY WARRANTY; without even the implied warranty of
- * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
- * GNU Affero General Public License for more details.
- *
- * You should have received a copy of the GNU Affero General Public License
- * along with this program.  If not, see <https://www.gnu.org/licenses/>.
- */
-
-use std::{collections::HashMap, io};
-
-use darkfi_serial::{Decodable, Encodable, SerialDecodable, SerialEncodable};
-
-use crate::{Error, Result};
-
-pub type Channel<T> = (smol::channel::Sender<T>, smol::channel::Receiver<T>);
-pub type Sender = (smol::channel::Sender<NetMsg>, smol::channel::Receiver<NetMsg>);
-
-#[derive(PartialEq, Eq, Debug, Clone)]
-pub enum Role {
-    Follower,
-    Candidate,
-    Leader,
-}
-
-#[derive(SerialDecodable, SerialEncodable, Clone, Debug)]
-pub struct SyncRequest {
-    pub id: u64,
-    pub logs_len: u64,
-    pub last_term: u64,
-}
-
-#[derive(SerialDecodable, SerialEncodable, Clone, Debug)]
-pub struct SyncResponse {
-    pub id: u64,
-    pub logs: Logs,
-    pub commit_length: u64,
-    pub leader_id: NodeId,
-    pub wipe: bool,
-}
-
-#[derive(SerialDecodable, SerialEncodable, Clone, Debug)]
-pub struct VoteRequest {
-    pub node_id: NodeId,
-    pub current_term: u64,
-    pub log_length: u64,
-    pub last_term: u64,
-}
-
-#[derive(SerialDecodable, SerialEncodable, Clone, Debug)]
-pub struct VoteResponse {
-    pub node_id: NodeId,
-    pub current_term: u64,
-    pub ok: bool,
-}
-
-#[derive(SerialDecodable, SerialEncodable, Clone, Debug)]
-pub struct LogRequest {
-    pub leader_id: NodeId,
-    pub current_term: u64,
-    pub prefix_len: u64,
-    pub prefix_term: u64,
-    pub commit_length: u64,
-    pub suffix: Logs,
-}
-
-#[derive(SerialDecodable, SerialEncodable, Clone, Debug)]
-pub struct LogResponse {
-    pub node_id: NodeId,
-    pub current_term: u64,
-    pub ack: u64,
-    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;
-    }
-}
-
-#[derive(SerialDecodable, SerialEncodable, Clone, Debug)]
-pub struct BroadcastMsgRequest(pub Vec<u8>);
-
-#[derive(Clone, Debug, SerialDecodable, SerialEncodable)]
-pub struct Log {
-    pub term: u64,
-    pub msg: Vec<u8>,
-}
-
-#[derive(Clone, Debug, Eq, PartialEq, Hash, SerialDecodable, SerialEncodable)]
-pub struct NodeId(pub String);
-
-#[derive(Clone, Debug, SerialDecodable, SerialEncodable)]
-pub struct Logs(pub Vec<Log>);
-
-impl Logs {
-    pub fn len(&self) -> u64 {
-        self.0.len() as u64
-    }
-    pub fn is_empty(&self) -> bool {
-        self.0.is_empty()
-    }
-
-    pub fn slice_from(&self, start: u64) -> Option<Self> {
-        if self.len() >= start {
-            return Some(Self(self.0[start as usize..].to_vec()))
-        }
-        None
-    }
-
-    pub fn slice_to(&self, end: u64) -> Self {
-        for i in (0..end).rev() {
-            if self.len() >= i {
-                return Self(self.0[..i as usize].to_vec())
-            }
-        }
-        Self(vec![])
-    }
-
-    pub fn get(&self, index: u64) -> Result<Log> {
-        match self.0.get(index as usize) {
-            Some(l) => Ok(l.clone()),
-            None => Err(Error::RaftError("unable to indexing into vector".into())),
-        }
-    }
-
-    pub fn to_vec(&self) -> Vec<Log> {
-        self.0.clone()
-    }
-}
-
-#[derive(Clone, Debug)]
-pub struct MapLength(pub HashMap<NodeId, u64>);
-
-impl MapLength {
-    pub fn get(&self, key: &NodeId) -> Result<u64> {
-        match self.0.get(key) {
-            Some(v) => Ok(*v),
-            None => Err(Error::RaftError("unable to indexing into HashMap".into())),
-        }
-    }
-
-    pub fn insert(&mut self, key: &NodeId, value: u64) {
-        self.0.insert(key.clone(), value);
-    }
-}
-
-#[derive(SerialDecodable, SerialEncodable, Clone, Debug)]
-pub struct NetMsg {
-    pub id: u64,
-    pub recipient_id: Option<NodeId>,
-    pub method: NetMsgMethod,
-    pub payload: Vec<u8>,
-}
-
-#[derive(Clone, Debug, PartialEq, Eq)]
-#[repr(u8)]
-pub enum NetMsgMethod {
-    LogResponse = 0,
-    LogRequest = 1,
-    VoteResponse = 2,
-    VoteRequest = 3,
-    BroadcastRequest = 4,
-    NodeIdMsg = 5,
-}
-
-impl Encodable for NetMsgMethod {
-    fn encode<S: io::Write>(&self, s: S) -> core::result::Result<usize, io::Error> {
-        let len: usize = match self {
-            Self::LogResponse => 0,
-            Self::LogRequest => 1,
-            Self::VoteResponse => 2,
-            Self::VoteRequest => 3,
-            Self::BroadcastRequest => 4,
-            Self::NodeIdMsg => 5,
-        };
-        (len as u8).encode(s)
-    }
-}
-
-impl Decodable for NetMsgMethod {
-    fn decode<D: io::Read>(d: D) -> core::result::Result<Self, io::Error> {
-        let com: u8 = Decodable::decode(d)?;
-        Ok(match com {
-            0 => Self::LogResponse,
-            1 => Self::LogRequest,
-            2 => Self::VoteResponse,
-            3 => Self::VoteRequest,
-            4 => Self::BroadcastRequest,
-            _ => Self::NodeIdMsg,
-        })
-    }
-}

src/raft/protocol_raft.rs

@@ -1,136 +0,0 @@
-/* This file is part of DarkFi (https://dark.fi)
- *
- * Copyright (C) 2020-2023 Dyne.org foundation
- *
- * This program is free software: you can redistribute it and/or modify
- * it under the terms of the GNU Affero General Public License as
- * published by the Free Software Foundation, either version 3 of the
- * License, or (at your option) any later version.
- *
- * This program is distributed in the hope that it will be useful,
- * but WITHOUT ANY WARRANTY; without even the implied warranty of
- * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
- * GNU Affero General Public License for more details.
- *
- * You should have received a copy of the GNU Affero General Public License
- * along with this program.  If not, see <https://www.gnu.org/licenses/>.
- */
-use std::collections::HashMap;
-
-use async_std::sync::{Arc, Mutex};
-use async_trait::async_trait;
-use chrono::Utc;
-use darkfi_serial::serialize;
-use log::debug;
-use rand::{rngs::OsRng, RngCore};
-use smol::Executor;
-
-use super::primitives::{NetMsg, NetMsgMethod, NodeId, NodeIdMsg};
-use crate::{net, Result};
-
-pub struct ProtocolRaft {
-    id: NodeId,
-    jobsman: net::ProtocolJobsManagerPtr,
-    notify_queue_sender: smol::channel::Sender<NetMsg>,
-    msg_sub: net::MessageSubscription<NetMsg>,
-    p2p: net::P2pPtr,
-    seen_msgs: Arc<Mutex<HashMap<String, i64>>>,
-    channel: net::ChannelPtr,
-}
-
-impl ProtocolRaft {
-    pub async fn init(
-        id: NodeId,
-        channel: net::ChannelPtr,
-        notify_queue_sender: smol::channel::Sender<NetMsg>,
-        p2p: net::P2pPtr,
-        seen_msgs: Arc<Mutex<HashMap<String, i64>>>,
-    ) -> net::ProtocolBasePtr {
-        let message_subsytem = channel.get_message_subsystem();
-        message_subsytem.add_dispatch::<NetMsg>().await;
-
-        let msg_sub = channel.subscribe_msg::<NetMsg>().await.expect("Missing NetMsg dispatcher!");
-
-        Arc::new(Self {
-            id,
-            notify_queue_sender,
-            msg_sub,
-            jobsman: net::ProtocolJobsManager::new("ProtocolRaft", channel.clone()),
-            p2p,
-            seen_msgs,
-            channel,
-        })
-    }
-
-    async fn handle_receive_msg(self: Arc<Self>) -> Result<()> {
-        debug!(target: "raft::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::protocol_raft",
-            "ProtocolRaft::handle_receive_msg() received id: {:?} method {:?}",
-            &msg.id, &msg.method
-            );
-
-            {
-                let mut msgs = self.seen_msgs.lock().await;
-                if msgs.contains_key(&msg.id.to_string()) {
-                    continue
-                }
-                msgs.insert(msg.id.to_string(), chrono::Utc::now().timestamp());
-            }
-
-            let msg = (*msg).clone();
-            self.p2p.broadcast(msg.clone()).await?;
-
-            // check if the local node and recipient id are equal
-            if let Some(recipient_id) = &msg.recipient_id {
-                if &self.id != recipient_id {
-                    continue
-                }
-            }
-
-            self.notify_queue_sender.send(msg).await?;
-        }
-    }
-}
-
-#[async_trait]
-impl net::ProtocolBase for ProtocolRaft {
-    /// Starts ping-pong keep-alive messages exchange. Runs ping-pong in the
-    /// 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::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::protocol_raft", "ProtocolRaft::start() [END]");
-        Ok(())
-    }
-
-    fn name(&self) -> &'static str {
-        "ProtocolRaft"
-    }
-}
-
-impl net::Message for NetMsg {
-    fn name() -> &'static str {
-        "netmsg"
-    }
-}

src/raft/settings.rs

@@ -1,49 +0,0 @@
-/* This file is part of DarkFi (https://dark.fi)
- *
- * Copyright (C) 2020-2023 Dyne.org foundation
- *
- * This program is free software: you can redistribute it and/or modify
- * it under the terms of the GNU Affero General Public License as
- * published by the Free Software Foundation, either version 3 of the
- * License, or (at your option) any later version.
- *
- * This program is distributed in the hope that it will be useful,
- * but WITHOUT ANY WARRANTY; without even the implied warranty of
- * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
- * GNU Affero General Public License for more details.
- *
- * You should have received a copy of the GNU Affero General Public License
- * along with this program.  If not, see <https://www.gnu.org/licenses/>.
- */
-
-use std::path::PathBuf;
-
-#[derive(Clone, Debug)]
-pub struct RaftSettings {
-    // the leader duration for sending heartbeat; in milliseconds
-    pub heartbeat_timeout: u64,
-
-    // the duration for electing new leader; in seconds
-    pub timeout: u64,
-
-    // the duration for sending id to other nodes; in seconds
-    pub id_timeout: u64,
-
-    // this duration used to clean up hashmaps; in seconds
-    pub prun_duration: i64,
-
-    // Datastore path
-    pub datastore_path: PathBuf,
-}
-
-impl Default for RaftSettings {
-    fn default() -> Self {
-        Self {
-            heartbeat_timeout: 500,
-            timeout: 6,
-            id_timeout: 12,
-            prun_duration: 30,
-            datastore_path: PathBuf::from(""),
-        }
-    }
-}