raft: avoid load all the logs from datasotre and implement slice_from and slice_to

src/raft/consensus.rs

@@ -286,9 +286,9 @@ impl<T: Decodable + Encodable + Clone> Raft<T> {
             let logs = if sr.logs_len == 0 {
                 self.logs()?.clone()
             } else if self.logs_len() >= sr.logs_len &&
-                self.logs()?.get(sr.logs_len - 1)?.term == sr.last_term
+                self.get_log(sr.logs_len - 1)?.term == sr.last_term
             {
-                self.logs()?.slice_from(sr.logs_len).unwrap()
+                self.slice_logs_from(sr.logs_len)?
             } else {
                 wipe = true;
                 self.logs()?.clone()
@@ -331,7 +331,7 @@ impl<T: Decodable + Encodable + Clone> Raft<T> {
         }
 
         for i in self.commits_len()..sr.commit_length {
-            self.push_commit(&self.logs()?.get(i)?.msg).await?;
+            self.push_commit(&self.get_log(i)?.msg).await?;
         }
 
         self.current_leader = Some(sr.leader_id.clone());
@@ -491,6 +491,7 @@ impl<T: Decodable + Encodable + Clone> Raft<T> {
         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,
@@ -500,16 +501,15 @@ impl<T: Decodable + Encodable + Clone> Raft<T> {
             }
         };
 
-        let suffix: Logs = if self.logs()?.slice_from(prefix_len).is_some() {
-            self.logs()?.slice_from(prefix_len).unwrap()
-        } else {
-            return Ok(())
+        let suffix: Logs = match self.slice_logs_from(prefix_len) {
+            Ok(l) => l,
+            Err(_) => return Ok(()),
         };
 
         let mut prefix_term = 0;
 
         if prefix_len > 0 {
-            prefix_term = self.logs()?.get(prefix_len - 1)?.term;
+            prefix_term = self.get_log(prefix_len - 1)?.term;
         }
 
         let request = LogRequest {
@@ -539,7 +539,7 @@ impl<T: Decodable + Encodable + Clone> Raft<T> {
         }
 
         let mut ok = (self.logs_len() >= lr.prefix_len) &&
-            (lr.prefix_len == 0 || self.logs()?.get(lr.prefix_len - 1)?.term == lr.prefix_term);
+            (lr.prefix_len == 0 || self.get_log(lr.prefix_len - 1)?.term == lr.prefix_term);
 
         let mut ack = 0;
 
@@ -626,10 +626,10 @@ impl<T: Decodable + Encodable + Clone> Raft<T> {
         let max_ready = *ready.iter().max().unwrap();
 
         if max_ready > self.commits_len() &&
-            self.logs()?.get(max_ready - 1)?.term == self.current_term()?
+            self.get_log(max_ready - 1)?.term == self.current_term()?
         {
             for i in self.commits_len()..max_ready {
-                self.push_commit(&self.logs()?.get(i)?.msg).await?;
+                self.push_commit(&self.get_log(i)?.msg).await?;
             }
         }
 
@@ -644,8 +644,8 @@ impl<T: Decodable + Encodable + Clone> Raft<T> {
     ) -> Result<()> {
         if !suffix.is_empty() && 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.push_logs(&self.logs()?.slice_to(prefix_len))?;
+            if self.get_log(index)?.term != suffix.get(index - prefix_len)?.term {
+                self.push_logs(&self.slice_logs_to(prefix_len)?)?;
             }
         }
 
@@ -657,7 +657,7 @@ impl<T: Decodable + Encodable + Clone> Raft<T> {
 
         if leader_commit > self.commits_len() {
             for i in self.commits_len()..leader_commit {
-                self.push_commit(&self.logs()?.get(i)?.msg).await?;
+                self.push_commit(&self.get_log(i)?.msg).await?;
             }
         }
 
@@ -703,4 +703,13 @@ impl<T: Decodable + Encodable + Clone> Raft<T> {
     fn last_log(&self) -> Result<Option<Log>> {
         self.datastore.logs.get_last()
     }
+    fn get_log(&self, index: u64) -> Result<Log> {
+        self.datastore.logs.get(index)
+    }
+    fn slice_logs_from(&self, index: u64) -> Result<Logs> {
+        Ok(Logs(self.datastore.logs.get_gt(index)?))
+    }
+    fn slice_logs_to(&self, index: u64) -> Result<Logs> {
+        Ok(Logs(self.datastore.logs.get_lt(index)?))
+    }
 }

src/raft/datastore.rs

@@ -5,7 +5,7 @@ use sled::Batch;
 
 use crate::{
     util::serial::{deserialize, serialize, Decodable, Encodable},
-    Result,
+    Error, Result,
 };
 
 use super::primitives::{Log, NodeId};
@@ -54,8 +54,12 @@ impl<T: Decodable + Encodable> DataTree<T> {
 
     pub fn insert(&self, data: &T) -> Result<()> {
         let serialized = serialize(data);
-        let datahash = blake3::hash(&serialized);
-        self.tree.insert(datahash.as_bytes(), serialized)?;
+        let last_index: u64 = if let Some(d) = self.tree.last()? {
+            u64::from_be_bytes(d.0.to_vec().try_into().unwrap())
+        } else {
+            0
+        };
+        self.tree.insert(last_index.to_be_bytes(), serialized)?;
         Ok(())
     }
 
@@ -64,10 +68,9 @@ impl<T: Decodable + Encodable> DataTree<T> {
 
         let mut batch = Batch::default();
 
-        for i in data {
-            let serialized = serialize(i);
-            let hash = blake3::hash(&serialized);
-            batch.insert(hash.as_bytes(), serialized);
+        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)?;
@@ -98,6 +101,53 @@ impl<T: Decodable + Encodable> DataTree<T> {
         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 get_lt(&self, index: u64) -> Result<Vec<T>> {
+        let mut ret: Vec<T> = Vec::new();
+
+        let index = index.to_be_bytes();
+
+        for i in self.tree.get_lt(index).iter() {
+            if i.is_none() {
+                return Ok(ret)
+            }
+            let data = i.as_ref().unwrap();
+            let data = deserialize(&data.1)?;
+            ret.push(data)
+        }
+
+        Ok(ret)
+    }
+
+    pub fn get_gt(&self, index: u64) -> Result<Vec<T>> {
+        let mut ret: Vec<T> = Vec::new();
+
+        let index = (index - 1).to_be_bytes();
+
+        for i in self.tree.get_gt(index).iter() {
+            if i.is_none() {
+                return Ok(ret)
+            }
+            let data = i.as_ref().unwrap();
+            let data = deserialize(&data.1)?;
+            ret.push(data)
+        }
+
+        Ok(ret)
+    }
+
     pub fn is_empty(&self) -> bool {
         self.tree.is_empty()
     }

src/raft/primitives.rs

@@ -104,22 +104,6 @@ impl Logs {
         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()),