feat: Implement log recovery

Author: michaelmior

ROADMAP.md

@@ -20,7 +20,7 @@ This stage will implement logging of all operations in ArgusDB to allow for faul
 
 - [x] Define a log file format in specs/logging.md
 - [x] Add log entries whenever new documents are inserted
-- [] Implement log recovery to read the log on restart and recover from a crash
+- [x] Implement log recovery to read the log on restart and recover from a crash
 - [] Implement log rotation and deletion of old log files
 
 # Stage 4 - Disk storage

specs/logging.md

@@ -9,14 +9,15 @@ Each log entry is a JSON object with the following fields:
 - `ts`: An ISO 8601 timestamp of when the operation occurred.
 - `op`: The type of operation. Can be one of "insert", "update", or "delete".
 - `doc`: The document for "insert" and "update" operations.
-- `id`: The document ID for "update" and "delete" operations.
+- `id`: The document ID for all operations.
 
 ### Insert operation
 
 ```json
 {
   "ts": "2026-01-10T12:00:00.000Z",
   "op": "insert",
+  "id": "01H4J3J4J3J4J3J4J3J4J3J4J3",
   "doc": { "a": 1 }
 }
 ```

src/log.rs

@@ -8,6 +8,7 @@ use std::path::Path;
 #[derive(Serialize, Deserialize, Debug)]
 pub enum Operation {
     Insert {
+        id: String,
         doc: Value,
     },
     Update {

src/storage.rs

@@ -13,47 +13,154 @@ pub struct MemTable {
 impl MemTable {
     pub fn new(log_path: &str) -> Self {
         let logger = Logger::new(log_path).unwrap();
-        MemTable {
+        let mut memtable = MemTable {
             documents: BTreeMap::new(),
             schema: Schema {
                 types: vec![],
                 properties: None,
                 items: None,
             },
             logger,
-        }
+        };
+        memtable.recover(log_path);
+        memtable
     }
 
-    pub fn insert(&mut self, doc: Value) -> String {
-        let doc_schema = infer_schema(&doc);
-        self.schema.merge(doc_schema.clone());
-        let id = Uuid::now_v7().to_string();
-        self.documents.insert(id.clone(), doc.clone());
-        self.logger
-            .log(Operation::Insert { doc })
-            .expect("Failed to log insert");
-        id
-    }
+        fn recover(&mut self, log_path: &str) {
 
-    pub fn delete(&mut self, id: &str) {
-        self.documents.remove(id);
-        self.logger
-            .log(Operation::Delete { id: id.to_string() })
-            .expect("Failed to log delete");
-    }
+            let log_content = std::fs::read_to_string(log_path).unwrap_or_default();
+
+            for line in log_content.lines() {
+
+                if line.is_empty() {
+
+                    continue;
+
+                }
+
+                let entry: crate::log::LogEntry = serde_json::from_str(line).unwrap();
+
+                match entry.op {
+
+                    Operation::Insert { id, doc } => {
+
+                        self.insert_with_id(&id, doc);
+
+                    }
+
+                    Operation::Update { id, doc } => {
+
+                        self._update(&id, doc);
+
+                    }
+
+                    Operation::Delete { id } => {
+
+                        self._delete(&id);
+
+                    }
+
+                }
+
+            }
+
+        }
+
+    
+
+        fn insert_with_id(&mut self, id: &str, doc: Value) {
+
+            let doc_schema = infer_schema(&doc);
+
+            self.schema.merge(doc_schema);
+
+            self.documents.insert(id.to_string(), doc);
+
+        }
+
+    
+
+        pub fn insert(&mut self, doc: Value) -> String {
+
+            let doc_schema = infer_schema(&doc);
+
+            self.schema.merge(doc_schema.clone());
+
+            let id = Uuid::now_v7().to_string();
+
+            self.documents.insert(id.clone(), doc.clone());
+
+            self.logger
+
+                .log(Operation::Insert {
+
+                    id: id.clone(),
+
+                    doc,
+
+                })
+
+                .expect("Failed to log insert");
+
+            id
+
+        }
+
+    
+
+        fn _delete(&mut self, id: &str) {
+
+            self.documents.remove(id);
+
+        }
+
+    
+
+        pub fn delete(&mut self, id: &str) {
+
+            self._delete(id);
+
+            self.logger
+
+                .log(Operation::Delete { id: id.to_string() })
+
+                .expect("Failed to log delete");
+
+        }
+
+    
+
+        fn _update(&mut self, id: &str, doc: Value) {
+
+            let doc_schema = infer_schema(&doc);
+
+            self.schema.merge(doc_schema);
+
+            self.documents.insert(id.to_string(), doc);
+
+        }
+
+    
+
+        pub fn update(&mut self, id: &str, doc: Value) {
+
+            self._update(id, doc.clone());
+
+            self.logger
+
+                .log(Operation::Update {
+
+                    id: id.to_string(),
+
+                    doc,
+
+                })
+
+                .expect("Failed to log update");
+
+        }
 
-    pub fn update(&mut self, id: &str, doc: Value) {
-        let doc_schema = infer_schema(&doc);
-        self.schema.merge(doc_schema);
-        self.documents.insert(id.to_string(), doc.clone());
-        self.logger
-            .log(Operation::Update {
-                id: id.to_string(),
-                doc,
-            })
-            .expect("Failed to log update");
     }
-}
 
 #[cfg(test)]
 mod tests {
@@ -125,7 +232,10 @@ mod tests {
 
         let entry1: crate::log::LogEntry = serde_json::from_str(lines.next().unwrap()).unwrap();
         match entry1.op {
-            Operation::Insert { doc } => assert_eq!(doc, doc1),
+            Operation::Insert { id, doc } => {
+                assert_eq!(id, id1);
+                assert_eq!(doc, doc1);
+            }
             _ => panic!("Expected insert operation"),
         }
 
@@ -144,4 +254,20 @@ mod tests {
             _ => panic!("Expected delete operation"),
         }
     }
+
+    #[test]
+    fn test_memtable_recover() {
+        let (log_file, mut memtable) = create_test_memtable();
+        let doc1 = json!({"a": 1});
+        let id1 = memtable.insert(doc1.clone());
+
+        let doc2 = json!({"b": "hello"});
+        let id2 = memtable.insert(doc2.clone());
+
+        memtable.delete(&id1);
+
+        let memtable2 = MemTable::new(log_file.path().to_str().unwrap());
+        assert_eq!(memtable2.documents.len(), 1);
+        assert_eq!(*memtable2.documents.get(&id2).unwrap(), doc2);
+    }
 }