fix: deregister failure detector in region migration (#4293)

* fix: deregister failure detector in region migration

* chore: apply suggestions from CR

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

@@ -245,7 +245,7 @@ impl Context {
     /// Notifies the RegionSupervisor to register failure detectors of failed region.
     ///
     /// The original failure detector was removed once the procedure was triggered.
-    /// Now, we need to register the failure detector for the failed region.
+    /// Now, we need to register the failure detector for the failed region again.
     pub async fn register_failure_detectors(&self) {
         let cluster_id = self.persistent_ctx.cluster_id;
         let datanode_id = self.persistent_ctx.from_peer.id;
@@ -256,6 +256,20 @@ impl Context {
             .await;
     }
 
+    /// Notifies the RegionSupervisor to deregister failure detectors.
+    ///
+    /// The original failure detectors was removed once the procedure was triggered.
+    /// However, the `from_peer` may still send the heartbeats contains the failed region.
+    pub async fn deregister_failure_detectors(&self) {
+        let cluster_id = self.persistent_ctx.cluster_id;
+        let datanode_id = self.persistent_ctx.from_peer.id;
+        let region_id = self.persistent_ctx.region_id;
+
+        self.region_failure_detector_controller
+            .deregister_failure_detectors(vec![(cluster_id, datanode_id, region_id)])
+            .await;
+    }
+
     /// Removes the `table_route` of [VolatileContext], returns true if any.
     pub fn remove_table_route_value(&mut self) -> bool {
         let value = self.volatile_ctx.table_route.take();

src/meta-srv/src/procedure/region_migration/update_metadata/upgrade_candidate_region.rs

@@ -175,6 +175,7 @@ impl UpdateMetadata {
         };
 
         ctx.remove_table_route_value();
+        ctx.deregister_failure_detectors().await;
         // Consumes the guard.
         ctx.volatile_ctx.opening_region_guard.take();