fix(meta-service): detach the SysData to avoid race condition (#18722)

* fix(meta-service): detach the SysData to avoid race condition

When creating a new level in state-machine, it should detach the SysData
to avoid race condition with snapshot building.

Before this commit, the new writable level and the snapshot compactor
shares the same data thus the new applied data increases the
`last-log-id` of a new built snapshot. Result in a snapshot that
lacks some log entries it declares to have.

* M  src/meta/raft-store/src/sm_v003/compact_immutable_levels_test.rs

Author: drmingdrmer

src/meta/raft-store/src/leveled_store/level.rs

@@ -108,7 +108,7 @@ impl Level {
     pub(crate) fn new_level(&self) -> Self {
         let level = Self {
             // sys data are cloned
-            sys_data: self.sys_data.clone(),
+            sys_data: Arc::new(Mutex::new(self.sys_data.lock().unwrap().clone())),
 
             // Large data set is referenced.
             kv: Default::default(),
@@ -147,3 +147,23 @@ impl Level {
         self.with_sys_data(|s| s.nodes_mut().clone())
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn test_new_level_detach_sys_data() {
+        let level1 = Level::default();
+        let level2 = level1.new_level();
+        level1.with_sys_data(|s: &mut SysData| {
+            s.update_seq(1);
+        });
+        assert_eq!(level1.with_sys_data(|s| s.curr_seq()), 1);
+
+        level2.with_sys_data(|s: &mut SysData| {
+            s.update_seq(10);
+        });
+        assert_eq!(level1.with_sys_data(|s| s.curr_seq()), 1);
+    }
+}

src/meta/raft-store/src/sm_v003/compact_immutable_levels_test.rs

@@ -150,7 +150,7 @@ async fn test_compact_3_level() -> anyhow::Result<()> {
 
     let (sys_data, strm) = compacting_data.compact_into_stream().await?;
     assert_eq!(
-        r#"{"last_applied":{"leader_id":{"term":3,"node_id":3},"index":3},"last_membership":{"log_id":{"leader_id":{"term":3,"node_id":3},"index":3},"membership":{"configs":[],"nodes":{}}},"nodes":{"3":{"name":"3","endpoint":{"addr":"3","port":3},"grpc_api_advertise_address":null}},"sequence":7,"data_seq":3}"#,
+        r#"{"last_applied":{"leader_id":{"term":3,"node_id":3},"index":3},"last_membership":{"log_id":{"leader_id":{"term":3,"node_id":3},"index":3},"membership":{"configs":[],"nodes":{}}},"nodes":{"3":{"name":"3","endpoint":{"addr":"3","port":3},"grpc_api_advertise_address":null}},"sequence":7,"data_seq":2}"#,
         serde_json::to_string(&sys_data).unwrap()
     );
 
@@ -182,7 +182,7 @@ async fn test_export_2_level_with_meta() -> anyhow::Result<()> {
         .await?;
 
     assert_eq!(
-        r#"{"last_applied":null,"last_membership":{"log_id":null,"membership":{"configs":[],"nodes":{}}},"nodes":{},"sequence":4,"data_seq":2}"#,
+        r#"{"last_applied":null,"last_membership":{"log_id":null,"membership":{"configs":[],"nodes":{}}},"nodes":{},"sequence":4,"data_seq":1}"#,
         serde_json::to_string(&sys_data).unwrap()
     );