feat: persistence with WAL

.github/workflows/rust.yml

@@ -21,5 +21,5 @@ jobs:
     - name: Build
       run: sudo apt-get install protobuf-compiler -y && cargo build --verbose
     - name: Run tests
-      run: cargo test --verbose
+      run: cargo test --workspace
 

Cargo.toml

@@ -14,16 +14,19 @@
 
 [workspace]
     members = [
+        "crates/persister",
         "crates/raft",
         "crates/timer",
     ]
     [workspace.dependencies]
         timer = {path = "crates/timer"}
         raft = {path = "crates/raft"}
+        persister = {path = "crates/persister"}
 
 [dependencies]
     timer.workspace = true
     raft.workspace = true
+    persister.workspace = true
     anyhow = "1.0.86"
     bytes = {version = "1.6.1", features = [
         "serde",

README.md

@@ -41,6 +41,7 @@ Then this update will be visible to all nodes:
 - [x] Automatic failover
 - [x] Read/Write splitting
 - [x] Write forwarding
+- [x] Persistence(WAL)
 - [ ] Log compaction(snapshot)
 - [ ] Dynamic membership
 

crates/persister/Cargo.toml

@@ -0,0 +1,14 @@
+[package]
+    name = "persister"
+    version = "0.1.0"
+    edition = "2021"
+
+[dependencies]
+    raft.workspace = true
+    anyhow = "1.0.86"
+    tokio = {version = "1.39.3", features = [
+        "rt-multi-thread",
+        "macros",
+        "fs",
+    ]}
+    async-trait = "0.1.81"

crates/persister/src/file.rs

@@ -0,0 +1,106 @@
+use super::{Persister, Term};
+use anyhow::Result;
+use async_trait::async_trait;
+use tokio::{
+    fs::{File, OpenOptions},
+    io::{AsyncReadExt, AsyncWriteExt},
+};
+
+pub struct FilePersister {
+    dir: String,
+
+    wal_w: Option<File>,
+    wal_r: Option<File>,
+}
+
+impl FilePersister {
+    pub fn new(dir: &str) -> Self {
+        Self {
+            dir: String::from(dir),
+            wal_w: None,
+            wal_r: None,
+        }
+    }
+
+    async fn wal_write_handle(&mut self) -> &mut File {
+        if self.wal_w.is_none() {
+            std::fs::create_dir_all(format!("{}/wal", self.dir)).unwrap();
+            let path = format!("{}/wal/log", self.dir);
+            let file = OpenOptions::new()
+                .create(true)
+                .write(true)
+                .open(path)
+                .await
+                .unwrap();
+            self.wal_w = Some(file);
+        }
+        self.wal_w.as_mut().unwrap()
+    }
+
+    async fn wal_read_handle(&mut self) -> std::io::Result<&mut File> {
+        if self.wal_r.is_none() {
+            let path = format!("{}/wal/log", self.dir);
+            let file = File::open(path).await?;
+            self.wal_r = Some(file);
+        }
+        Ok(self.wal_r.as_mut().unwrap())
+    }
+}
+
+#[async_trait]
+impl Persister for FilePersister {
+    async fn read_wal(&mut self) -> Result<Option<(Term, Vec<u8>)>> {
+        let wal = match self.wal_read_handle().await {
+            Ok(wal) => wal,
+            Err(e) => {
+                if e.kind() == std::io::ErrorKind::NotFound {
+                    return Ok(None);
+                } else {
+                    return Err(e.into());
+                }
+            }
+        };
+
+        // Read the length of the entry
+        let mut length_bytes = [0u8; size_of::<usize>()];
+        match wal.read_exact(&mut length_bytes).await {
+            Ok(_) => {}
+            Err(e) if e.kind() == std::io::ErrorKind::UnexpectedEof => {
+                // Close current wal file so next call can read it again
+                self.wal_r = None;
+                return Ok(None);
+            }
+            Err(e) => return Err(e.into()),
+        }
+        let length = usize::from_le_bytes(length_bytes);
+
+        // Read the term
+        let mut term_bytes = [0u8; size_of::<Term>()];
+        wal.read_exact(&mut term_bytes).await?;
+        let term = Term::from_le_bytes(term_bytes);
+
+        // Read the data
+        let mut data = vec![0u8; length - size_of::<Term>()];
+        wal.read_exact(&mut data).await?;
+
+        Ok(Some((term, data)))
+    }
+
+    async fn write_wal(&mut self, term: Term, data: &[u8]) -> Result<()> {
+        let wal = self.wal_write_handle().await;
+        let length = size_of::<Term>() + data.len();
+        wal.write(&length.to_le_bytes()).await?;
+        wal.write(&term.to_le_bytes()).await?;
+        wal.write(data).await?;
+        Ok(())
+    }
+
+    async fn read_snapshot(&self) -> Result<Option<(usize, Vec<u8>)>> {
+        // TODO: implement me
+        Ok(None)
+    }
+    async fn write_snapshot(&mut self, _last_index: usize, _data: &[u8]) -> Result<()> {
+        // TODO: implement me
+        Ok(())
+    }
+}

crates/persister/src/lib.rs

@@ -0,0 +1,6 @@
+pub use raft::state::Term;
+pub use raft::Persister;
+
+mod file;
+
+pub use file::FilePersister;

crates/persister/tests/file_test.rs

@@ -0,0 +1,45 @@
+use persister::FilePersister;
+use persister::Persister;
+
+#[tokio::test]
+async fn wal_not_exist() {
+    let mut fp = fp_at("not_exist");
+    assert_eq!(None, fp.read_wal().await.unwrap());
+}
+
+#[tokio::test]
+async fn rw_wal() {
+    let mut fp = fp_at("file");
+
+    let write_wal = vec![
+        (1, "wal_test".as_bytes()),
+        (2, "wal_test2".as_bytes()),
+        (3, "wal_test3".as_bytes()),
+    ];
+    for (term, data) in write_wal.iter() {
+        fp.write_wal(*term, *data).await.unwrap();
+    }
+
+    for (term, data) in write_wal.iter() {
+        let (t, d) = fp.read_wal().await.unwrap().unwrap();
+        assert_eq!(t, *term);
+        assert_eq!(d.as_slice(), *data);
+    }
+
+    assert_eq!(None, fp.read_wal().await.unwrap());
+
+    // When meet EOF, next `read_wal` should start from the beginning
+    for (term, data) in write_wal.iter() {
+        let (t, d) = fp.read_wal().await.unwrap().unwrap();
+        assert_eq!(t, *term);
+        assert_eq!(d.as_slice(), *data);
+    }
+
+    assert_eq!(None, fp.read_wal().await.unwrap());
+}
+
+////////////////////////////////////////////////////////////////////////////////
+
+fn fp_at(dir: &str) -> FilePersister {
+    FilePersister::new(format!("/tmp/radis/test/persister/{}", dir).as_str())
+}

crates/raft/Cargo.toml

@@ -22,6 +22,7 @@
         "rt-multi-thread",
     ]}
     tonic = "0.12.1"
+    async-trait = "0.1.81"
 
 [build-dependencies]
     tonic-build = "0.11.0"

crates/raft/src/context.rs

@@ -2,6 +2,8 @@ use super::config::Config;
 use super::config::REQUEST_TIMEOUT;
 use super::log::LogManager;
 use super::service::PeerClient;
+use super::Persister;
+use anyhow::Result;
 use log::debug;
 use std::collections::HashMap;
 use std::sync::Arc;
@@ -22,7 +24,6 @@ pub struct Context {
     log: LogManager,
     peer_next_index: Vec<Arc<Mutex<LogIndex>>>,
     peer_sync_index: Vec<Arc<RwLock<LogIndex>>>,
-    commit_ch: mpsc::Sender<Arc<Vec<u8>>>,
 
     timeout: Arc<OneshotTimer>,
     tick: Arc<PeriodicTimer>,
@@ -61,16 +62,19 @@ impl Context {
             id,
             peers,
             id_map: Arc::new(id_map),
-            log: LogManager::new(),
+            log: LogManager::new(commit_ch),
             peer_next_index: (0..n_peer).map(|_| Arc::new(Mutex::new(0))).collect(),
             peer_sync_index: (0..n_peer).map(|_| Arc::new(RwLock::new(0))).collect(),
-            commit_ch,
 
             timeout: Arc::new(OneshotTimer::new(timeout_event)),
             tick: Arc::new(PeriodicTimer::new(tick_event)),
         }
     }
 
+    pub async fn setup_persister(&mut self, persister: Box<dyn Persister>) -> Result<()> {
+        self.log.setup_persister(persister).await
+    }
+
     pub async fn init_timer(&self) {
         let timeout = self.timeout.clone();
         let tick = self.tick.clone();
@@ -129,7 +133,7 @@ impl Context {
     }
 
     pub async fn commit_log(&mut self, index: LogIndex) {
-        self.log.commit(index, &self.commit_ch).await;
+        self.log.commit(index).await;
     }
 
     pub fn peer_next_index(&self, peer: Peer) -> Arc<Mutex<LogIndex>> {

crates/raft/src/lib.rs

@@ -8,4 +8,5 @@ mod log;
 mod service;
 pub mod state;
 
+pub use log::Persister;
 pub use service::RaftService;