sync prepare/rollback merge

Signed-off-by: 5kbpers <[email protected]>
5kbpers · Oct 13, 2022 · 32200fe · 32200fe
1 parent ae2c430
commit 32200fe
Show file tree

Hide file tree

Showing 2 changed files with 19 additions and 8 deletions.
diff --git a/components/raftstore/src/store/recover.rs b/components/raftstore/src/store/recover.rs
@@ -365,8 +365,7 @@ impl<EK: KvEngine, ER: RaftEngine> Recovery<EK, ER> {
         let target_peer = target_region
             .get_peers()
             .iter()
-            .filter(|p| p.store_id == store_id)
-            .next()
+            .find(|p| p.store_id == store_id)
             .cloned()
             .unwrap();
         let _ = self.router.force_send(

diff --git a/components/raftstore/src/store/worker/seqno_relation.rs b/components/raftstore/src/store/worker/seqno_relation.rs
@@ -109,7 +109,9 @@ impl<S: Snapshot> fmt::Display for Task<S> {
 }
 
 enum HandleExecResResult {
-    RegionStateChanged(RegionLocalState),
+    ConfChange(RegionLocalState),
+    PrepareMerge(RegionLocalState),
+    RollbackMerge(RegionLocalState),
     SplitRegion {
         new_created_regions: Vec<RegionLocalState>,
         region_state: RegionLocalState,
@@ -196,8 +198,8 @@ impl<EK: KvEngine, ER: RaftEngine> Runner<EK, ER> {
 
             for result in self.handle_exec_res(res.region_id, seqno.get_number(), &res.exec_res) {
                 match result {
-                    HandleExecResResult::RegionStateChanged(region_state) => {
-                        info!("region state changed"; "region_id" => res.region_id, "region_state" => ?region_state, "apply_state" => ?res.apply_state);
+                    HandleExecResResult::ConfChange(region_state) => {
+                        info!("region conf changed"; "region_id" => res.region_id, "region_state" => ?region_state, "apply_state" => ?res.apply_state);
                         relation.set_region_state(region_state);
                     }
                     HandleExecResResult::SplitRegion {
@@ -226,6 +228,16 @@ impl<EK: KvEngine, ER: RaftEngine> Runner<EK, ER> {
                         // restart.
                         sync_relation_with_current_batch.push(res.region_id);
                     }
+                    HandleExecResResult::PrepareMerge(region_state) => {
+                        info!("prepare merge"; "region_id" => res.region_id, "region_state" => ?region_state, "apply_state" => ?res.apply_state);
+                        relation.set_region_state(region_state);
+                        sync_relation_with_current_batch.push(res.region_id);
+                    }
+                    HandleExecResResult::RollbackMerge(region_state) => {
+                        info!("rollback merge"; "region_id" => res.region_id, "region_state" => ?region_state, "apply_state" => ?res.apply_state);
+                        relation.set_region_state(region_state);
+                        sync_relation_with_current_batch.push(res.region_id);
+                    }
                     HandleExecResResult::CommitMerge {
                         target_state,
                         source_relation,
@@ -420,7 +432,7 @@ impl<EK: KvEngine, ER: RaftEngine> Runner<EK, ER> {
                         let mut state = RegionLocalState::default();
                         state.set_region(cp.region.clone());
                         state.set_state(PeerState::Normal);
-                        results.push(HandleExecResResult::RegionStateChanged(state));
+                        results.push(HandleExecResResult::ConfChange(state));
                     };
                 }
                 ExecResult::SplitRegion {
@@ -458,7 +470,7 @@ impl<EK: KvEngine, ER: RaftEngine> Runner<EK, ER> {
                     state.set_region(region.clone());
                     state.set_state(PeerState::Merging);
                     state.set_merge_state(merge_state.clone());
-                    results.push(HandleExecResResult::RegionStateChanged(state));
+                    results.push(HandleExecResResult::PrepareMerge(state));
                 }
                 ExecResult::CommitMerge {
                     region,
@@ -500,7 +512,7 @@ impl<EK: KvEngine, ER: RaftEngine> Runner<EK, ER> {
                     let mut state = RegionLocalState::default();
                     state.set_region(region.clone());
                     state.set_state(PeerState::Normal);
-                    results.push(HandleExecResResult::RegionStateChanged(state));
+                    results.push(HandleExecResResult::RollbackMerge(state));
                 }
                 ExecResult::DeleteRange { .. }
                 | ExecResult::IngestSst { .. }