fix: remove stale region failover detectors (#6221)

* fix: remove stale region failover detectors

* fix: fix unit tests

src/meta-srv/src/error.rs

@@ -69,6 +69,18 @@ pub enum Error {
         region_id: RegionId,
     },
 
+    #[snafu(display(
+        "The region migration procedure is completed for region: {}, target_peer: {}",
+        region_id,
+        target_peer_id
+    ))]
+    RegionMigrated {
+        #[snafu(implicit)]
+        location: Location,
+        region_id: RegionId,
+        target_peer_id: u64,
+    },
+
     #[snafu(display("The region migration procedure aborted, reason: {}", reason))]
     MigrationAbort {
         #[snafu(implicit)]
@@ -953,7 +965,8 @@ impl ErrorExt for Error {
             | Error::RegionOpeningRace { .. }
             | Error::RegionRouteNotFound { .. }
             | Error::MigrationAbort { .. }
-            | Error::MigrationRunning { .. } => StatusCode::Unexpected,
+            | Error::MigrationRunning { .. }
+            | Error::RegionMigrated { .. } => StatusCode::Unexpected,
             Error::TableNotFound { .. } => StatusCode::TableNotFound,
             Error::SaveClusterInfo { source, .. }
             | Error::InvalidClusterInfoFormat { source, .. }

src/meta-srv/src/procedure/region_migration/manager.rs

@@ -326,7 +326,11 @@ impl RegionMigrationManager {
 
         if self.has_migrated(&region_route, &task)? {
             info!("Skipping region migration task: {task}");
-            return Ok(None);
+            return error::RegionMigratedSnafu {
+                region_id,
+                target_peer_id: task.to_peer.id,
+            }
+            .fail();
         }
 
         self.verify_region_leader_peer(&region_route, &task)?;
@@ -567,7 +571,8 @@ mod test {
         env.create_physical_table_metadata(table_info, region_routes)
             .await;
 
-        manager.submit_procedure(task).await.unwrap();
+        let err = manager.submit_procedure(task).await.unwrap_err();
+        assert_matches!(err, error::Error::RegionMigrated { .. });
     }
 
     #[tokio::test]

src/meta-srv/src/region/supervisor.rs

@@ -27,7 +27,7 @@ use common_meta::DatanodeId;
 use common_runtime::JoinHandle;
 use common_telemetry::{debug, error, info, warn};
 use common_time::util::current_time_millis;
-use error::Error::{LeaderPeerChanged, MigrationRunning, TableRouteNotFound};
+use error::Error::{LeaderPeerChanged, MigrationRunning, RegionMigrated, TableRouteNotFound};
 use snafu::{ensure, OptionExt, ResultExt};
 use store_api::storage::RegionId;
 use tokio::sync::mpsc::{Receiver, Sender};
@@ -527,6 +527,7 @@ impl RegionSupervisor {
 
     async fn do_failover(&mut self, task: RegionMigrationProcedureTask, count: u32) -> Result<()> {
         let from_peer_id = task.from_peer.id;
+        let to_peer_id = task.to_peer.id;
         let region_id = task.region_id;
 
         info!(
@@ -536,6 +537,15 @@ impl RegionSupervisor {
 
         if let Err(err) = self.region_migration_manager.submit_procedure(task).await {
             return match err {
+                RegionMigrated { .. } => {
+                    info!(
+                        "Region has been migrated to target peer: {}, removed failover detector for region: {}, datanode: {}",
+                        to_peer_id, region_id, from_peer_id
+                    );
+                    self.deregister_failure_detectors(vec![(from_peer_id, region_id)])
+                        .await;
+                    Ok(())
+                }
                 // Returns Ok if it's running or table is dropped.
                 MigrationRunning { .. } => {
                     info!(

tests-integration/tests/region_migration.rs

@@ -32,6 +32,7 @@ use datatypes::vectors::{Helper, UInt64Vector};
 use frontend::error::Result as FrontendResult;
 use frontend::instance::Instance;
 use futures::future::BoxFuture;
+use meta_srv::error;
 use meta_srv::error::Result as MetaResult;
 use meta_srv::metasrv::SelectorContext;
 use meta_srv::procedure::region_migration::RegionMigrationProcedureTask;
@@ -213,7 +214,7 @@ pub async fn test_region_migration(store_type: StorageType, endpoints: Vec<Strin
     assert_values(cluster.fe_instance()).await;
 
     // Triggers again.
-    let procedure = region_migration_manager
+    let err = region_migration_manager
         .submit_procedure(RegionMigrationProcedureTask::new(
             region_id,
             peer_factory(from_peer_id),
@@ -221,8 +222,8 @@ pub async fn test_region_migration(store_type: StorageType, endpoints: Vec<Strin
             Duration::from_millis(1000),
         ))
         .await
-        .unwrap();
-    assert!(procedure.is_none());
+        .unwrap_err();
+    assert!(matches!(err, error::Error::RegionMigrated { .. }));
 }
 
 /// A naive metric table region migration test by SQL function
@@ -471,7 +472,7 @@ pub async fn test_region_migration_by_sql(store_type: StorageType, endpoints: Ve
     assert_values(cluster.fe_instance()).await;
 
     // Triggers again.
-    let procedure = region_migration_manager
+    let err = region_migration_manager
         .submit_procedure(RegionMigrationProcedureTask::new(
             region_id,
             peer_factory(from_peer_id),
@@ -479,8 +480,8 @@ pub async fn test_region_migration_by_sql(store_type: StorageType, endpoints: Ve
             Duration::from_millis(1000),
         ))
         .await
-        .unwrap();
-    assert!(procedure.is_none());
+        .unwrap_err();
+    assert!(matches!(err, error::Error::RegionMigrated { .. }));
 
     let new_distribution = find_region_distribution_by_sql(&cluster, TEST_TABLE_NAME).await;
 
@@ -621,7 +622,7 @@ pub async fn test_region_migration_multiple_regions(
     assert_values(cluster.fe_instance()).await;
 
     // Triggers again.
-    let procedure = region_migration_manager
+    let err = region_migration_manager
         .submit_procedure(RegionMigrationProcedureTask::new(
             region_id,
             peer_factory(from_peer_id),
@@ -629,8 +630,8 @@ pub async fn test_region_migration_multiple_regions(
             Duration::from_millis(1000),
         ))
         .await
-        .unwrap();
-    assert!(procedure.is_none());
+        .unwrap_err();
+    assert!(matches!(err, error::Error::RegionMigrated { .. }));
 }
 
 /// A region migration test for a region server contains all regions of the table.
@@ -757,7 +758,7 @@ pub async fn test_region_migration_all_regions(store_type: StorageType, endpoint
     assert_values(cluster.fe_instance()).await;
 
     // Triggers again.
-    let procedure = region_migration_manager
+    let err = region_migration_manager
         .submit_procedure(RegionMigrationProcedureTask::new(
             region_id,
             peer_factory(from_peer_id),
@@ -765,8 +766,8 @@ pub async fn test_region_migration_all_regions(store_type: StorageType, endpoint
             Duration::from_millis(1000),
         ))
         .await
-        .unwrap();
-    assert!(procedure.is_none());
+        .unwrap_err();
+    assert!(matches!(err, error::Error::RegionMigrated { .. }));
 }
 
 pub async fn test_region_migration_incorrect_from_peer(