script/research/raft_p2p: integrate Sled and save logs in local storage

script/research/raft_p2p/Cargo.lock

@@ -19,6 +19,18 @@ version = "1.0.56"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "4361135be9122e0870de935d7c439aef945b9f9ddd4199a553b5270b49c82a27"
 
+[[package]]
+name = "arrayref"
+version = "0.3.6"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "a4c527152e37cf757a3f78aae5a06fbeefdb07ccc535c980a3208ee3060dd544"
+
+[[package]]
+name = "arrayvec"
+version = "0.7.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "8da52d66c7071e2e3fa2a1e5c6d088fec47b593032b254f5e980de8ea54454d6"
+
 [[package]]
 name = "async-attributes"
 version = "1.1.2"
@@ -260,6 +272,20 @@ version = "1.3.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "bef38d45163c2f1dde094a7dfd33ccf595c92905c8f8f4fdc18d06fb1037718a"
 
+[[package]]
+name = "blake3"
+version = "1.3.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "a08e53fc5a564bb15bfe6fae56bd71522205f1f91893f9c0116edad6496c183f"
+dependencies = [
+ "arrayref",
+ "arrayvec",
+ "cc",
+ "cfg-if",
+ "constant_time_eq",
+ "digest",
+]
+
 [[package]]
 name = "block-buffer"
 version = "0.10.2"
@@ -403,6 +429,12 @@ dependencies = [
  "cache-padded",
 ]
 
+[[package]]
+name = "constant_time_eq"
+version = "0.1.5"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "245097e9a4535ee1e3e3931fcfcd55a796a44c643e8596ff6566d68f09b87bbc"
+
 [[package]]
 name = "core-foundation"
 version = "0.9.3"
@@ -428,6 +460,29 @@ dependencies = [
  "libc",
 ]
 
+[[package]]
+name = "crc32fast"
+version = "1.3.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "b540bd8bc810d3885c6ea91e2018302f68baba2129ab3e88f32389ee9370880d"
+dependencies = [
+ "cfg-if",
+]
+
+[[package]]
+name = "crossbeam-epoch"
+version = "0.9.8"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "1145cf131a2c6ba0615079ab6a638f7e1973ac9c2634fcbeaaad6114246efe8c"
+dependencies = [
+ "autocfg",
+ "cfg-if",
+ "crossbeam-utils",
+ "lazy_static",
+ "memoffset",
+ "scopeguard",
+]
+
 [[package]]
 name = "crossbeam-utils"
 version = "0.8.8"
@@ -529,6 +584,7 @@ checksum = "f2fb860ca6fafa5552fb6d0e816a69c8e49f0908bf524e30a90d97c85892d506"
 dependencies = [
  "block-buffer",
  "crypto-common",
+ "subtle",
 ]
 
 [[package]]
@@ -625,6 +681,16 @@ dependencies = [
  "percent-encoding",
 ]
 
+[[package]]
+name = "fs2"
+version = "0.4.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "9564fc758e15025b46aa6643b1b77d047d1a56a1aea6e01002ac0c7026876213"
+dependencies = [
+ "libc",
+ "winapi",
+]
+
 [[package]]
 name = "futures"
 version = "0.3.21"
@@ -895,6 +961,16 @@ version = "0.2.121"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "efaa7b300f3b5fe8eb6bf21ce3895e1751d9665086af2d64b42f19701015ff4f"
 
+[[package]]
+name = "lock_api"
+version = "0.4.7"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "327fa5b6a6940e4699ec49a9beae1ea4845c6bab9314e4f84ac68742139d8c53"
+dependencies = [
+ "autocfg",
+ "scopeguard",
+]
+
 [[package]]
 name = "log"
 version = "0.4.16"
@@ -917,6 +993,15 @@ version = "2.4.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "308cc39be01b73d0d18f82a0e7b2a3df85245f84af96fdddc5d202d27e47b86a"
 
+[[package]]
+name = "memoffset"
+version = "0.6.5"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "5aa361d4faea93603064a027415f07bd8e1d5c88c9fbf68bf56a285428fd79ce"
+dependencies = [
+ "autocfg",
+]
+
 [[package]]
 name = "native-tls"
 version = "0.2.10"
@@ -1038,6 +1123,31 @@ version = "2.0.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "427c3892f9e783d91cc128285287e70a59e206ca452770ece88a76f7a3eddd72"
 
+[[package]]
+name = "parking_lot"
+version = "0.11.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "7d17b78036a60663b797adeaee46f5c9dfebb86948d1255007a1d6be0271ff99"
+dependencies = [
+ "instant",
+ "lock_api",
+ "parking_lot_core",
+]
+
+[[package]]
+name = "parking_lot_core"
+version = "0.8.5"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "d76e8e1493bcac0d2766c42737f34458f1c8c50c0d23bcb24ea953affb273216"
+dependencies = [
+ "cfg-if",
+ "instant",
+ "libc",
+ "redox_syscall",
+ "smallvec",
+ "winapi",
+]
+
 [[package]]
 name = "pem"
 version = "1.0.2"
@@ -1159,6 +1269,7 @@ dependencies = [
  "async-executor",
  "async-std",
  "async-trait",
+ "blake3",
  "borsh",
  "clap",
  "darkfi",
@@ -1167,6 +1278,7 @@ dependencies = [
  "log",
  "rand",
  "simplelog",
+ "sled",
  "smol",
 ]
 
@@ -1293,6 +1405,12 @@ dependencies = [
  "winapi",
 ]
 
+[[package]]
+name = "scopeguard"
+version = "1.1.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "d29ab0c6d3fc0ee92fe66e2d99f700eab17a8d57d1c1d3b748380fb20baa78cd"
+
 [[package]]
 name = "sct"
 version = "0.7.0"
@@ -1404,6 +1522,28 @@ version = "0.4.6"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "eb703cfe953bccee95685111adeedb76fabe4e97549a58d16f03ea7b9367bb32"
 
+[[package]]
+name = "sled"
+version = "0.34.7"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "7f96b4737c2ce5987354855aed3797279def4ebf734436c6aa4552cf8e169935"
+dependencies = [
+ "crc32fast",
+ "crossbeam-epoch",
+ "crossbeam-utils",
+ "fs2",
+ "fxhash",
+ "libc",
+ "log",
+ "parking_lot",
+]
+
+[[package]]
+name = "smallvec"
+version = "1.8.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "f2dd574626839106c320a323308629dcb1acfc96e32a8cba364ddc61ac23ee83"
+
 [[package]]
 name = "smol"
 version = "1.2.5"
@@ -1444,6 +1584,12 @@ version = "0.10.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "73473c0e59e6d5812c5dfe2a064a6444949f089e20eec9a2e5506596494e4623"
 
+[[package]]
+name = "subtle"
+version = "2.4.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "6bdef32e8150c2a081110b42772ffe7d7c9032b606bc226c8260fd97e0976601"
+
 [[package]]
 name = "syn"
 version = "1.0.91"

script/research/raft_p2p/Cargo.toml

@@ -23,6 +23,10 @@ borsh = "0.9.3"
 log = "0.4.16"
 rand = "0.8.5"
 
+# database
+sled = "0.34.7"
+blake3 = "1.3.1"
+
 [dev-dependencies]
 clap = {version = "3.1.6", features = ["derive"]}
 easy-parallel = "3.2.0"

script/research/raft_p2p/src/datastore.rs

@@ -0,0 +1,92 @@
+use std::marker::PhantomData;
+
+use sled::Batch;
+
+use darkfi::{
+    util::serial::{deserialize, serialize, Decodable, Encodable},
+    Result,
+};
+
+use crate::{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";
+
+pub struct DataStore {
+    _db: sled::Db,
+    pub logs: DataTree<Log>,
+    pub commits: DataTree<Vec<u8>>,
+    pub commits_length: DataTree<u64>,
+    pub voted_for: DataTree<Option<NodeId>>,
+    pub current_term: DataTree<u64>,
+}
+
+impl DataStore {
+    pub fn new(db_path: &str) -> Result<Self> {
+        let _db = sled::open(db_path).unwrap();
+        let logs = DataTree::new(&_db, SLED_LOGS_TREE)?;
+        let commits = DataTree::new(&_db, SLED_COMMITS_TREE)?;
+        let commits_length = DataTree::new(&_db, SLED_COMMITS_LENGTH_TREE)?;
+        let voted_for = DataTree::new(&_db, SLED_VOTED_FOR_TREE)?;
+        let current_term = DataTree::new(&_db, SLED_CURRENT_TERM_TREE)?;
+
+        Ok(Self { _db, logs, commits, commits_length, voted_for, current_term })
+    }
+}
+
+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).unwrap();
+        Ok(Self { tree, phantom: PhantomData })
+    }
+
+    pub fn insert(&self, data: &T) -> Result<()> {
+        let serialized = serialize(data);
+        let datahash = blake3::hash(&serialized);
+        self.tree.insert(datahash.as_bytes(), serialized).unwrap();
+        Ok(())
+    }
+
+    pub fn wipe_insert_all(&self, data: &Vec<T>) -> Result<()> {
+        self.tree.clear().unwrap();
+
+        let mut batch = Batch::default();
+
+        for i in data {
+            let serialized = serialize(i);
+            let hash = blake3::hash(&serialized);
+            batch.insert(hash.as_bytes(), serialized);
+        }
+
+        self.tree.apply_batch(batch).unwrap();
+
+        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.unwrap().1)?;
+            ret.push(da)
+        }
+
+        Ok(ret)
+    }
+
+    pub fn get_last(&self) -> Result<Option<T>> {
+        if let Some(found) = self.tree.last().unwrap() {
+            let da = deserialize(&found.1).unwrap();
+            return Ok(Some(da))
+        }
+        Ok(None)
+    }
+}

script/research/raft_p2p/src/lib.rs

@@ -1,12 +1,14 @@
-use std::io;
+use std::{io, net::SocketAddr};
 
 use borsh::{BorshDeserialize, BorshSerialize};
 
-use darkfi::util::serial::{Decodable, Encodable, SerialDecodable, SerialEncodable};
+use darkfi::util::serial::{serialize, Decodable, Encodable, SerialDecodable, SerialEncodable};
 
+pub mod datastore;
 pub mod p2p;
 pub mod raft;
 
+pub use datastore::DataStore;
 pub use p2p::ProtocolRaft;
 
 #[derive(PartialEq, Eq)]
@@ -16,12 +18,7 @@ pub enum Role {
     Leader,
 }
 
-impl Default for Role {
-    fn default() -> Self {
-        Self::Follower
-    }
-}
-#[derive(BorshSerialize, BorshDeserialize, Clone, Debug, Default)]
+#[derive(BorshSerialize, BorshDeserialize, Clone, Debug)]
 pub struct VoteRequest {
     node_id: NodeId,
     current_term: u64,
@@ -29,14 +26,14 @@ pub struct VoteRequest {
     last_term: u64,
 }
 
-#[derive(BorshSerialize, BorshDeserialize, Clone, Debug, Default)]
+#[derive(BorshSerialize, BorshDeserialize, Clone, Debug)]
 pub struct VoteResponse {
     node_id: NodeId,
     current_term: u64,
     ok: bool,
 }
 
-#[derive(BorshSerialize, BorshDeserialize, Clone, Debug, Default)]
+#[derive(BorshSerialize, BorshDeserialize, Clone, Debug)]
 pub struct LogRequest {
     leader_id: NodeId,
     current_term: u64,
@@ -46,7 +43,7 @@ pub struct LogRequest {
     suffix: VecR<Log>,
 }
 
-#[derive(BorshSerialize, BorshDeserialize, Clone, Debug, Default)]
+#[derive(BorshSerialize, BorshDeserialize, Clone, Debug)]
 pub struct LogResponse {
     node_id: NodeId,
     current_term: u64,
@@ -60,7 +57,7 @@ impl VoteResponse {
     }
 }
 
-#[derive(BorshSerialize, BorshDeserialize, Clone, Debug, Default)]
+#[derive(BorshSerialize, BorshDeserialize, Clone, Debug, SerialDecodable, SerialEncodable)]
 pub struct Log {
     term: u64,
     msg: Vec<u8>,
@@ -71,16 +68,23 @@ pub struct Log {
     BorshDeserialize,
     Clone,
     Debug,
-    Default,
     Eq,
     PartialEq,
     Hash,
     SerialDecodable,
     SerialEncodable,
 )]
-pub struct NodeId(pub u64);
+pub struct NodeId(pub Vec<u8>);
 
-#[derive(BorshSerialize, BorshDeserialize, Clone, Debug, Default)]
+impl From<SocketAddr> for NodeId {
+    fn from(addr: SocketAddr) -> Self {
+        let ser = serialize(&addr);
+        let hash = blake3::hash(&ser).as_bytes().to_vec();
+        Self(hash)
+    }
+}
+
+#[derive(BorshSerialize, BorshDeserialize, Clone, Debug)]
 pub struct VecR<T: BorshSerialize + BorshDeserialize>(pub Vec<T>);
 
 impl<T: BorshSerialize + BorshDeserialize + Clone> VecR<T> {
@@ -102,6 +106,10 @@ impl<T: BorshSerialize + BorshDeserialize + Clone> VecR<T> {
     pub fn get(&self, index: u64) -> T {
         self.0[index as usize].clone()
     }
+
+    pub fn to_vec(&self) -> Vec<T> {
+        self.0[..].to_vec()
+    }
 }
 
 #[derive(

script/research/raft_p2p/src/raft.rs

@@ -1,34 +1,36 @@
-use async_std::{sync::Arc, task};
-use std::{cmp::min, collections::HashMap, time::Duration};
+use async_std::{
+    sync::{Arc, Mutex},
+    task,
+};
+use std::{cmp::min, collections::HashMap, net::SocketAddr, path::PathBuf, time::Duration};
 
 use async_executor::Executor;
 use borsh::BorshSerialize;
 use futures::{select, FutureExt};
+use log::error;
 use rand::Rng;
 
 use darkfi::{net, Result};
 
 use crate::{
-    try_from_slice_unchecked, Log, LogRequest, LogResponse, NetMsg, NetMsgMethod, NodeId,
-    ProtocolRaft, Role, VecR, VoteRequest, VoteResponse,
+    try_from_slice_unchecked, DataStore, Log, LogRequest, LogResponse, NetMsg, NetMsgMethod,
+    NodeId, ProtocolRaft, Role, VecR, VoteRequest, VoteResponse,
 };
 
 const HEARTBEATTIMEOUT: u64 = 100;
 const TIMEOUT: u64 = 300;
 
-#[derive(Default)]
 pub struct Raft {
     // this will be derived from the ip
     id: NodeId,
 
-    // these four vars should be on local storage
+    // these five vars should be on local storage
     current_term: u64,
     voted_for: Option<NodeId>,
     logs: VecR<Log>,
     commit_length: u64,
-
     // the log will be added to this vector if it's committed by the majority of nodes
-    commits: Vec<Vec<u8>>,
+    commits: Arc<Mutex<Vec<Vec<u8>>>>,
 
     role: Role,
 
@@ -44,11 +46,54 @@ pub struct Raft {
     last_term: u64,
 
     send_queues: Option<async_channel::Sender<NetMsg>>,
+
+    datastore: DataStore,
 }
 
 impl Raft {
-    pub fn new() -> Self {
-        Self::default()
+    pub fn new(addr: SocketAddr, db_path: PathBuf) -> Result<Self> {
+        if db_path.to_str().is_none() {
+            error!(target: "raft", "datastore path is incorrect");
+            return Err(darkfi::Error::ParseFailed("unable to parse pathbuf to str"))
+        };
+
+        let db_path_str = db_path.to_str().unwrap();
+
+        let mut current_term = 0;
+        let mut voted_for = None;
+        let mut logs = VecR(vec![]);
+        let mut commit_length = 0;
+        let mut commits = Arc::new(Mutex::new(vec![]));
+
+        let datastore = if db_path.exists() {
+            let datastore = DataStore::new(db_path_str)?;
+            current_term = datastore.current_term.get_last()?.unwrap_or(0);
+            voted_for = datastore.voted_for.get_last()?.flatten();
+            logs = VecR(datastore.logs.get_all()?);
+            commit_length = datastore.commits_length.get_last()?.unwrap_or(0);
+            commits = Arc::new(Mutex::new(datastore.commits.get_all()?));
+            datastore
+        } else {
+            DataStore::new(db_path_str)?
+        };
+
+        Ok(Self {
+            id: NodeId::from(addr),
+            current_term,
+            voted_for,
+            logs,
+            commit_length,
+            commits,
+            role: Role::Follower,
+            current_leader: None,
+            votes_received: VecR(vec![]),
+            sent_length: HashMap::new(),
+            acked_length: HashMap::new(),
+            nodes: VecR(vec![]),
+            last_term: 0,
+            send_queues: None,
+            datastore,
+        })
     }
 
     pub async fn start(
@@ -112,22 +157,15 @@ impl Raft {
         }
     }
 
-    pub fn recover_from_local_storage(
-        current_term: u64,
-        voted_for: Option<NodeId>,
-        commit_length: u64,
-        logs: VecR<Log>,
-    ) -> Self {
-        Self { current_term, voted_for, commit_length, logs, ..Default::default() }
-    }
-
-    pub fn get_commits(&self) -> Vec<Vec<u8>> {
+    pub fn get_commits(&self) -> Arc<Mutex<Vec<Vec<u8>>>> {
         self.commits.clone()
     }
 
     pub async fn broadcast_msg(&mut self, msg: Vec<u8>) {
         if self.role == Role::Leader {
-            self.logs.push(&Log { msg, term: self.current_term });
+            let log = Log { msg, term: self.current_term };
+            self.push_log(&log).unwrap();
+
             self.acked_length.insert(self.id.clone(), self.logs.len());
 
             let nodes = self.nodes.0.clone();
@@ -176,9 +214,9 @@ impl Raft {
     }
 
     async fn send_vote_request(&mut self) {
-        self.current_term += 1;
+        self.set_current_term(&(self.current_term + 1)).unwrap();
         self.role = Role::Candidate;
-        self.voted_for = Some(self.id.clone());
+        self.set_voted_for(&Some(self.id.clone())).unwrap();
         self.votes_received.push(&self.id);
 
         self.reset_last_term();
@@ -196,8 +234,8 @@ impl Raft {
 
     async fn receive_vote_request(&mut self, vr: VoteRequest) {
         if vr.current_term > self.current_term {
-            self.current_term = vr.current_term;
-            self.voted_for = None;
+            self.set_current_term(&vr.current_term).unwrap();
+            self.set_voted_for(&None).unwrap();
             self.role = Role::Follower;
         }
 
@@ -218,7 +256,7 @@ impl Raft {
             VoteResponse { node_id: self.id.clone(), current_term: self.current_term, ok: false };
 
         if vr.current_term == self.current_term && vote_ok && vote {
-            self.voted_for = Some(vr.node_id.clone());
+            self.set_voted_for(&Some(vr.node_id.clone())).unwrap();
             response.set_ok(true);
         }
 
@@ -240,9 +278,9 @@ impl Raft {
                 }
             }
         } else if vr.current_term > self.current_term {
-            self.current_term = vr.current_term;
+            self.set_current_term(&vr.current_term).unwrap();
             self.role = Role::Follower;
-            self.voted_for = None;
+            self.set_voted_for(&None).unwrap();
         }
     }
 
@@ -270,8 +308,8 @@ impl Raft {
 
     async fn receive_log_request(&mut self, lr: LogRequest) {
         if lr.current_term > self.current_term {
-            self.current_term = lr.current_term;
-            self.voted_for = None;
+            self.set_current_term(&lr.current_term).unwrap();
+            self.set_voted_for(&None).unwrap();
         }
 
         if lr.current_term == self.current_term {
@@ -283,7 +321,7 @@ impl Raft {
             (lr.prefix_len == 0 || self.logs.get(lr.prefix_len - 1).term == lr.prefix_term);
 
         let response: LogResponse = if lr.current_term == self.current_term && ok {
-            self.append_log(lr.prefix_len, lr.commit_length, &lr.suffix);
+            self.append_log(lr.prefix_len, lr.commit_length, &lr.suffix).await;
             let ack = lr.prefix_len + lr.suffix.len();
             LogResponse { node_id: self.id.clone(), current_term: self.current_term, ack, ok }
         } else {
@@ -304,15 +342,15 @@ impl Raft {
             if lr.ok && lr.ack >= self.acked_length[&lr.node_id] {
                 self.sent_length.insert(lr.node_id.clone(), lr.ack);
                 self.acked_length.insert(lr.node_id, lr.ack);
-                self.commit_log();
+                self.commit_log().await;
             } else if self.sent_length[&lr.node_id] > 0 {
                 self.sent_length.insert(lr.node_id.clone(), self.sent_length[&lr.node_id] - 1);
                 self.update_logs(&lr.node_id).await;
             }
         } else if lr.current_term > self.current_term {
-            self.current_term = lr.current_term;
+            self.set_current_term(&lr.current_term).unwrap();
             self.role = Role::Follower;
-            self.voted_for = None;
+            self.set_voted_for(&None).unwrap();
         }
     }
 
@@ -330,7 +368,7 @@ impl Raft {
         )
     }
 
-    fn commit_log(&mut self) {
+    async fn commit_log(&mut self) {
         let min_acks = (self.nodes.len() + 1) / 2;
 
         let ready: Vec<u64> = self
@@ -350,32 +388,57 @@ impl Raft {
         if max_ready > self.commit_length && self.logs.get(max_ready - 1).term == self.current_term
         {
             for i in self.commit_length..(max_ready - 1) {
-                self.commits.push(self.logs.get(i).msg);
+                self.push_commit(&self.logs.get(i).msg).await.unwrap();
             }
 
-            self.commit_length = max_ready;
+            self.set_commit_length(&max_ready).unwrap();
         }
     }
 
-    fn append_log(&mut self, prefix_len: u64, leader_commit: u64, suffix: &VecR<Log>) {
+    async fn append_log(&mut self, prefix_len: u64, leader_commit: u64, suffix: &VecR<Log>) {
         if suffix.len() > 0 && self.logs.len() > prefix_len {
             let index = min(self.logs.len(), prefix_len + suffix.len()) - 1;
             if self.logs.get(index).term != suffix.get(index - prefix_len).term {
-                self.logs = self.logs.slice_to(prefix_len - 1);
+                self.push_logs(&self.logs.slice_to(prefix_len - 1)).unwrap();
             }
         }
 
         if prefix_len + suffix.len() > self.logs.len() {
             for i in (self.logs.len() - prefix_len)..(suffix.len() - 1) {
-                self.logs.push(&suffix.get(i));
+                self.push_log(&suffix.get(i)).unwrap();
             }
         }
 
         if leader_commit > self.commit_length {
             for i in self.commit_length..(leader_commit - 1) {
-                self.commits.push(self.logs.get(i).msg);
+                self.push_commit(&self.logs.get(i).msg).await.unwrap();
             }
-            self.commit_length = leader_commit;
+            self.set_commit_length(&leader_commit).unwrap();
         }
     }
+
+    fn set_commit_length(&mut self, i: &u64) -> Result<()> {
+        self.commit_length = *i;
+        self.datastore.commits_length.insert(i)
+    }
+    fn set_current_term(&mut self, i: &u64) -> Result<()> {
+        self.current_term = *i;
+        self.datastore.current_term.insert(i)
+    }
+    fn set_voted_for(&mut self, i: &Option<NodeId>) -> Result<()> {
+        self.voted_for = i.clone();
+        self.datastore.voted_for.insert(i)
+    }
+    async fn push_commit(&mut self, i: &Vec<u8>) -> Result<()> {
+        self.commits.lock().await.push(i.clone());
+        self.datastore.commits.insert(i)
+    }
+    fn push_log(&mut self, i: &Log) -> Result<()> {
+        self.logs.push(i);
+        self.datastore.logs.insert(i)
+    }
+    fn push_logs(&mut self, i: &VecR<Log>) -> Result<()> {
+        self.logs = i.clone();
+        self.datastore.logs.wipe_insert_all(&i.to_vec())
+    }
 }