fix: remark region as inactive on leader changed (#2446)

* fix: remark reigon as inactive on leader changed

* chore: by comment

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

@@ -270,6 +270,8 @@ trait State: Sync + Send + Debug {
     fn status(&self) -> Status {
         Status::executing(true)
     }
+
+    fn remark_inactive_region_if_needed(&mut self) {}
 }
 
 /// The states transition of region failover procedure:
@@ -339,7 +341,11 @@ impl RegionFailoverProcedure {
     }
 
     fn from_json(json: &str, context: RegionFailoverContext) -> ProcedureResult<Self> {
-        let node: Node = serde_json::from_str(json).context(FromJsonSnafu)?;
+        let mut node: Node = serde_json::from_str(json).context(FromJsonSnafu)?;
+        // If the meta leader node dies during the execution of the procedure,
+        // the new leader node needs to remark the failed region as "inactive"
+        // to prevent it from renewing the lease.
+        node.state.remark_inactive_region_if_needed();
         Ok(Self { node, context })
     }
 }

src/meta-srv/src/procedure/region_failover/activate_region.rs

@@ -37,13 +37,18 @@ use crate::service::mailbox::{Channel, MailboxReceiver};
 #[derive(Serialize, Deserialize, Debug)]
 pub(super) struct ActivateRegion {
     candidate: Peer,
+    // If the meta leader node dies during the execution of the procedure,
+    // the new leader node needs to remark the failed region as "inactive"
+    // to prevent it from renewing the lease.
+    remark_inactive_region: bool,
     region_storage_path: Option<String>,
 }
 
 impl ActivateRegion {
     pub(super) fn new(candidate: Peer) -> Self {
         Self {
             candidate,
+            remark_inactive_region: false,
             region_storage_path: None,
         }
     }
@@ -55,7 +60,6 @@ impl ActivateRegion {
         timeout: Duration,
     ) -> Result<MailboxReceiver> {
         let table_id = failed_region.table_id;
-        // TODO(weny): considers fetching table info only once.
         let table_info = ctx
             .table_metadata_manager
             .table_info_manager()
@@ -168,12 +172,23 @@ impl State for ActivateRegion {
         ctx: &RegionFailoverContext,
         failed_region: &RegionIdent,
     ) -> Result<Box<dyn State>> {
+        if self.remark_inactive_region {
+            // Remark the fail region as inactive to prevent it from renewing the lease.
+            InactiveRegionManager::new(&ctx.in_memory)
+                .register_inactive_region(failed_region)
+                .await?;
+        }
+
         let mailbox_receiver = self
             .send_open_region_message(ctx, failed_region, OPEN_REGION_MESSAGE_TIMEOUT)
             .await?;
 
         self.handle_response(mailbox_receiver, failed_region).await
     }
+
+    fn remark_inactive_region_if_needed(&mut self) {
+        self.remark_inactive_region = true;
+    }
 }
 
 #[cfg(test)]

src/meta-srv/src/procedure/region_failover/deactivate_region.rs

@@ -226,7 +226,7 @@ mod tests {
             .unwrap();
         assert_eq!(
             format!("{next_state:?}"),
-            r#"ActivateRegion { candidate: Peer { id: 2, addr: "" }, region_storage_path: None }"#
+            r#"ActivateRegion { candidate: Peer { id: 2, addr: "" }, remark_inactive_region: false, region_storage_path: None }"#
         );
     }
 
@@ -268,7 +268,7 @@ mod tests {
         // Timeout or not, proceed to `ActivateRegion`.
         assert_eq!(
             format!("{next_state:?}"),
-            r#"ActivateRegion { candidate: Peer { id: 2, addr: "" }, region_storage_path: None }"#
+            r#"ActivateRegion { candidate: Peer { id: 2, addr: "" }, remark_inactive_region: false, region_storage_path: None }"#
         );
     }
 }