feat: Disallow to overwrite compacted raft logs (#128)

* feat: add compare first index when appending the log entries

* fix

* chore: fix formatting

---------

Co-authored-by: Gyubong <jopemachine@naver.com>

raftify/src/storage/heed_storage/mod.rs

@@ -1,6 +1,13 @@
 mod codec;
 mod constant;
 
+use self::codec::{format_entry_key_string, HeedEntry, HeedEntryKeyString};
+use super::{utils::append_compacted_logs, StableStorage, StorageType};
+use crate::{
+    config::Config,
+    error::Result,
+    raft::{self, prelude::*, GetEntriesContext},
+};
 use bincode::{deserialize, serialize};
 use constant::{CONF_STATE_KEY, HARD_STATE_KEY, LAST_INDEX_KEY, SNAPSHOT_KEY};
 use heed::{
@@ -17,14 +24,6 @@ use std::{
     sync::Arc,
 };
 
-use self::codec::{format_entry_key_string, HeedEntry, HeedEntryKeyString};
-use super::{utils::append_compacted_logs, StableStorage, StorageType};
-use crate::{
-    config::Config,
-    error::Result,
-    raft::{self, prelude::*, GetEntriesContext},
-};
-
 #[derive(Clone)]
 pub struct HeedStorage(Arc<RwLock<HeedStorageCore>>);
 
@@ -474,6 +473,16 @@ impl HeedStorageCore {
             return Ok(());
         }
 
+        let first_index = self.first_index(writer)?;
+
+        if first_index > entries[0].index {
+            self.logger.fatal(&format!(
+                "overwrite compacted raft logs, compacted: {}, append: {}",
+                first_index - 1,
+                entries[0].index,
+            ));
+        }
+
         let mut last_index = self.last_index(writer)?;
 
         if last_index + 1 < entries[0].index {
@@ -882,10 +891,10 @@ mod test {
             ),
             // TODO: Support the below test cases
             // overwrite compacted raft logs is not allowed
-            // (
-            //     vec![new_entry(2, 3), new_entry(3, 3), new_entry(4, 5)],
-            //     None,
-            // ),
+            (
+                vec![new_entry(2, 3), new_entry(3, 3), new_entry(4, 5)],
+                None,
+            ),
             // truncate the existing entries and append
             // (
             //     vec![new_entry(4, 5)],