fix: improve snapshot restore safety, Raft hard-state durability, and inode nlink atomicity

- Use hard-link copy (RocksUtils::link_dir) instead of rename in DBEngine::restore so
  the source checkpoint is preserved after restore
- Explicitly drop the old DB handle via a temporary path before any filesystem
  operations to release the RocksDB LOCK reliably
- Flush memtable + WAL before creating a checkpoint to guarantee all in-memory data
  is captured in SST files
- Fix set_hard_state_commit to actually persist the commit index to RocksDB
  (previously it only updated the in-memory field), and use commit_sync so the
  write is fsync-ed before returning
- Fix purge_checkpoint to never delete the currently active checkpoint: only
  consider directories whose mtime is strictly older than the current checkpoint
- Fix increment_inode_nlink / decrement_inode_nlink to accept the caller's
  WriteBatch so that nlink changes and edge mutations are committed atomically
- Fix unlink operations to correctly propagate DeleteResult (block list) when
  nlink reaches zero

Author: bigbigxu

curvine-common/src/raft/raft_node.rs

@@ -183,6 +183,8 @@ where
         app_store: &B,
         voters: Vec<u64>,
     ) -> RaftResult<u64> {
+        info!("init raft state: {:?}", log_store.initial_state()?);
+
         let spend = TimeSpent::new();
 
         match log_store.latest_snapshot()? {

curvine-common/src/raft/storage/rocks_storage_core.rs

@@ -105,13 +105,21 @@ impl RocksStorageCore {
 
     pub fn set_hard_state(&mut self, hs: HardState) -> RaftResult<()> {
         self.raft_state.hard_state = hs.clone();
-        self.db
-            .put_cf(Self::CF_META, Self::STATE_KEY, hs.encode_to_vec())?;
+
+        let mut batch = StoreWriteBatch::new(&self.db);
+        batch.set_state(&hs)?;
+        batch.commit()?;
+
         Ok(())
     }
 
     pub fn set_hard_state_commit(&mut self, commit: u64) -> RaftResult<()> {
         self.mut_hard_state().set_commit(commit);
+
+        let mut batch = StoreWriteBatch::new(&self.db);
+        batch.set_state(&self.raft_state.hard_state)?;
+        batch.commit()?;
+
         Ok(())
     }
 
@@ -414,7 +422,16 @@ impl<'a> StoreWriteBatch<'a> {
         Ok(())
     }
 
+    fn set_state(&mut self, state: &HardState) -> CommonResult<()> {
+        self.0.put_cf(
+            RocksStorageCore::CF_META,
+            RocksStorageCore::STATE_KEY,
+            state.encode_to_vec(),
+        )?;
+        Ok(())
+    }
+
     fn commit(self) -> CommonResult<()> {
-        self.0.commit()
+        self.0.commit_sync()
     }
 }

curvine-common/src/rocksdb/db_engine.rs

@@ -55,21 +55,19 @@ impl DBEngine {
     }
 
     pub fn restore<T: AsRef<str>>(&mut self, checkpoint: T) -> CommonResult<()> {
-        let db_path = &self.conf.data_dir;
+        let db_path = self.conf.data_dir.clone();
         let db_opt = self.conf.create_db_opt();
         let cfs = self.conf.create_cf_opt();
+        let checkpoint = checkpoint.as_ref();
 
         //The database points to a temporary directory.
         let tmp_path = Utils::temp_file();
         self.db = try_err!(DB::open(&db_opt, &tmp_path));
+        let _ = FileUtils::delete_path(&tmp_path, true);
+        FileUtils::delete_path(&db_path, true)?;
 
-        // Delete the original file and move the checkpoint to the data directory.
-        FileUtils::delete_path(db_path, true)?;
-        FileUtils::rename(checkpoint.as_ref(), db_path)?;
-
-        // Retry instantiating db and delete the temporary directory.
-        self.db = try_err!(DB::open_cf_with_opts(&db_opt, db_path, cfs));
-        let _ = FileUtils::delete_path(tmp_path, true);
+        try_err!(RocksUtils::link_dir(checkpoint, &db_path));
+        self.db = try_err!(DB::open_cf_with_opts(&db_opt, &db_path, cfs));
 
         Ok(())
     }
@@ -204,6 +202,8 @@ impl DBEngine {
 
     // Create a checkpoint.
     pub fn create_checkpoint(&self, id: u64) -> CommonResult<String> {
+        self.flush(true)?;
+
         let checkpoint_path = self.get_checkpoint_path(id);
         let existed = FileUtils::exists(&checkpoint_path);
 
@@ -246,8 +246,10 @@ impl DBEngine {
         &self.db
     }
 
-    pub fn flush(&self) -> CommonResult<()> {
-        self.db.flush()?;
+    pub fn flush_mem(&self, sync: bool) -> CommonResult<()> {
+        let mut opts = FlushOptions::default();
+        opts.set_wait(sync);
+        self.db.flush_opt(&opts)?;
         Ok(())
     }
 
@@ -256,6 +258,14 @@ impl DBEngine {
         Ok(())
     }
 
+    pub fn flush(&self, sync: bool) -> CommonResult<()> {
+        self.flush_mem(sync)?;
+        if !self.conf.disable_wal {
+            self.flush_wal(sync)?;
+        }
+        Ok(())
+    }
+
     pub fn write_batch(&self, batch: WriteBatchWithTransaction<false>) -> CommonResult<()> {
         try_err!(self.db.write(batch));
         Ok(())

curvine-common/src/rocksdb/rocks_utils.rs

@@ -12,9 +12,11 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+use std::fs;
 use byteorder::{BigEndian, ByteOrder};
 use orpc::{err_box, CommonResult};
 use prost::bytes::BufMut;
+use orpc::io::IOResult;
 
 // A utility class that converts some types to bytes.
 pub struct RocksUtils;
@@ -180,6 +182,35 @@ impl RocksUtils {
 
         end
     }
+
+    pub fn link_dir<P: AsRef<std::path::Path>>(src: P, dst: P) -> IOResult<()> {
+        let src = src.as_ref();
+        let dst = dst.as_ref();
+        if !dst.exists() {
+            fs::create_dir_all(dst)?;
+        }
+
+        for entry in fs::read_dir(src)? {
+            let entry = entry?;
+            let src_path = entry.path();
+            let dst_path = dst.join(entry.file_name());
+            if src_path.is_dir() {
+                Self::link_dir(&src_path, &dst_path)?;
+            } else if Self::is_sst_file(&src_path) {
+                fs::hard_link(&src_path, &dst_path)?;
+            } else {
+                fs::copy(&src_path, &dst_path)?;
+            }
+        }
+        Ok(())
+    }
+
+    fn is_sst_file(path: &std::path::Path) -> bool {
+        matches!(
+            path.extension().and_then(|e: &std::ffi::OsStr| e.to_str()),
+            Some("sst") | Some("ldb")
+        )
+    }
 }
 
 #[cfg(test)]

curvine-common/src/rocksdb/write_batch.rs

@@ -52,9 +52,13 @@ impl<'a> WriteBatch<'a> {
         self.db.write_batch(self.batch)
     }
 
-    pub fn commit_wal(self, sync: bool) -> CommonResult<()> {
+    pub fn commit_flush(self, sync: bool) -> CommonResult<()> {
         self.db.write_batch(self.batch)?;
-        self.db.flush_wal(sync)?;
+        self.db.flush(sync)?;
         Ok(())
     }
+
+    pub fn commit_sync(self) -> CommonResult<()> {
+        self.commit_flush(true)
+    }
 }

curvine-server/src/master/journal/journal_loader.rs

@@ -646,26 +646,37 @@ impl JournalLoader {
 
     // Clean up expired checkpoints.
     pub fn purge_checkpoint(&self, current_ck: impl AsRef<str>) -> CommonResult<()> {
-        let ck_dir = match Path::new(current_ck.as_ref()).parent() {
+        let current_ck = current_ck.as_ref();
+        let ck_dir = match Path::new(current_ck).parent() {
             None => return Ok(()),
             Some(v) => v,
         };
 
+        let current_mtime = match Path::new(current_ck).metadata() {
+            Ok(meta) => FileUtils::mtime(&meta)?,
+            Err(_) => return Ok(()),
+        };
+
         let mut vec = vec![];
         for entry in fs::read_dir(ck_dir)? {
             let entry = entry?;
             let meta = entry.metadata()?;
-            vec.push((FileUtils::mtime(&meta)?, entry.path()));
+            let mtime = FileUtils::mtime(&meta)?;
+            if mtime < current_mtime {
+                vec.push((mtime, entry.path()));
+            }
         }
 
-        // Sort by modification time
+        // Sort oldest-first and keep at most (retain_checkpoint_num - 1) older
+        // checkpoints so that together with current_ck the total is retain_checkpoint_num.
         vec.sort_by_key(|x| x.0);
-        let del_num = vec.len().saturating_sub(self.retain_checkpoint_num);
+        let keep = self.retain_checkpoint_num.saturating_sub(1);
+        let del_num = vec.len().saturating_sub(keep);
 
         for i in 0..del_num {
             let path = vec[i].1.as_path();
             FileUtils::delete_path(path, true)?;
-            info!("delete expired checkpoint, dir: {}", path.to_string_lossy())
+            info!("delete expired checkpoint: {}", path.to_string_lossy());
         }
 
         Ok(())

curvine-server/src/master/meta/fs_dir.rs

@@ -27,7 +27,7 @@ use curvine_common::state::{
     MkdirOpts, MountInfo, RenameFlags, SetAttrOpts, WorkerAddress,
 };
 use curvine_common::FsResult;
-use log::{info, warn};
+use log::{debug, info, warn};
 use orpc::common::{LocalTime, TimeSpent};
 use orpc::sync::AtomicCounter;
 use orpc::{err_box, err_ext, try_option, CommonResult};
@@ -94,7 +94,9 @@ impl FsDir {
     }
 
     pub fn update_op_id(&self, op_id: u64) {
-        self.op_id.set(op_id);
+        if op_id > self.op_id.get() {
+            self.op_id.set(op_id);
+        }
     }
 
     pub fn get_ttl_bucket_list(&self) -> Arc<TtlBucketList> {
@@ -362,7 +364,7 @@ impl FsDir {
         let name = inp.name().to_string();
 
         // Create an inode file node.
-        let file = InodeFile::with_opts(self.inode_id.next()?, LocalTime::mills() as i64, opts);
+        let file = InodeFile::with_opts(self.next_inode_id()?, LocalTime::mills() as i64, opts);
         inp = self.add_last_inode(inp, File(name, file))?;
         self.journal_writer.log_create_file(self, &inp)?;
 
@@ -582,9 +584,9 @@ impl FsDir {
                     Ok(true)
                 } else {
                     err_box!(
-                        "block_id {} resolves to inode_id {} which is not a file",
+                        "block_id {} resolves to inode {:?} which is not a file",
                         block_id,
-                        file_id
+                        v
                     )
                 }
             }
@@ -676,7 +678,7 @@ impl FsDir {
         let mut spend = TimeSpent::new();
         let path = path.as_ref();
 
-        // Set to other values ​​first to facilitate memory recycling.
+        // Set to other value first to facilitate memory recycling.
         self.root_dir = Self::create_root();
 
         // Reset rocksdb
@@ -691,10 +693,10 @@ impl FsDir {
         let time2 = spend.used_ms();
 
         info!(
-            "Restore from {}, restore rocksdb used {} ms, \
+            "restore from {}, restore rocksdb used {} ms, \
         build in-memory directory tree used {} ms, \
-        statistics updated during tree reconstruction",
-            path, time1, time2
+        statistics updated during tree reconstruction, last_inode_id {}",
+            path, time1, time2, last_inode_id
         );
         Ok(())
     }
@@ -985,7 +987,7 @@ impl FsDir {
             return Ok(DeleteResult::new());
         }
         let del_blocks = file.resize(opts.clone())?;
-        info!("resize file {} success, opts: {:?}", inp.path(), opts);
+        debug!("resize file {} success, opts: {:?}", inp.path(), opts);
 
         file.complete(file.len, &[], "", true)?;
         let mut del_res = DeleteResult::new();

curvine-server/src/master/meta/inode/inodes_children.rs

@@ -153,8 +153,8 @@ impl InodeChildren {
                         // But return owned pointer to complete object
                         Ok(InodePtr::from_owned(*inode))
                     } else {
-                        // Directory directly stores complete object and returns its reference
-                        let inserted = v.insert(inode.clone());
+                        // Directory: move ownership into the map and return a reference.
+                        let inserted = v.insert(inode);
                         Ok(InodePtr::from_ref(inserted.as_ref()))
                     }
                 }