core/lib: recover mvcc logical log if needed on connect

core/lib.rs

@@ -403,7 +403,8 @@ impl Database {
         let mv_store = if opts.enable_mvcc {
             let file = io.open_file(&format!("{path}-lg"), OpenFlags::default(), false)?;
             let storage = mvcc::persistent_storage::Storage::new(file);
-            Some(Arc::new(MvStore::new(mvcc::LocalClock::new(), storage)))
+            let mv_store = MvStore::new(mvcc::LocalClock::new(), storage);
+            Some(Arc::new(mv_store))
         } else {
             None
         };
@@ -482,6 +483,11 @@ impl Database {
     #[instrument(skip_all, level = Level::INFO)]
     pub fn connect(self: &Arc<Database>) -> Result<Arc<Connection>> {
         let pager = self.init_pager(None)?;
+        let pager = Arc::new(pager);
+
+        if self.mv_store.is_some() {
+            self.maybe_recover_logical_log(pager.clone())?;
+        }
 
         let page_size = pager.get_page_size_unchecked();
 
@@ -492,7 +498,7 @@ impl Database {
             .get();
         let conn = Arc::new(Connection {
             db: self.clone(),
-            pager: RwLock::new(Arc::new(pager)),
+            pager: RwLock::new(pager),
             schema: RwLock::new(
                 self.schema
                     .lock()
@@ -532,6 +538,18 @@ impl Database {
         Ok(conn)
     }
 
+    pub fn maybe_recover_logical_log(self: &Arc<Database>, pager: Arc<Pager>) -> Result<()> {
+        let Some(mv_store) = self.mv_store.clone() else {
+            panic!("tryign to recover logical log without mvcc");
+
+        };
+        if !mv_store.needs_recover() {
+            return Ok(());
+        }
+
+        mv_store.recover_logical_log(&self.io, &pager)
+    }
+
     pub fn is_readonly(&self) -> bool {
         self.open_flags.contains(OpenFlags::ReadOnly)
     }