fix: fix campaign reset not clearing leader state-s

WenyXu · WenyXu · commit f5e009a861aa · 2025-11-10T10:58:19.000Z
Signed-off-by: WenyXu &lt;wenymedia@gmail.com&gt;
diff --git a/src/meta-srv/src/election/rds/mysql.rs b/src/meta-srv/src/election/rds/mysql.rs
@@ -17,7 +17,7 @@ use std::sync::atomic::{AtomicBool, Ordering};
 use std::time::Duration;
 
 use common_meta::key::{CANDIDATES_ROOT, ELECTION_KEY};
-use common_telemetry::{error, warn};
+use common_telemetry::{error, info, warn};
 use common_time::Timestamp;
 use snafu::{OptionExt, ResultExt, ensure};
 use sqlx::mysql::{MySqlArguments, MySqlRow};
@@ -645,6 +645,13 @@ impl Election for MySqlElection {
     }
 
     async fn reset_campaign(&self) {
+        info!("Resetting campaign");
+        if self.is_leader.load(Ordering::Relaxed) {
+            if let Err(err) = self.step_down_without_lock().await {
+                error!(err; "Failed to step down without lock");
+            }
+            info!("Step down without lock successfully, due to reset campaign");
+        }
         if let Err(err) = self.client.lock().await.reset_client().await {
             error!(err; "Failed to reset client");
         }
diff --git a/src/meta-srv/src/election/rds/postgres.rs b/src/meta-srv/src/election/rds/postgres.rs
@@ -17,7 +17,7 @@ use std::sync::atomic::{AtomicBool, Ordering};
 use std::time::Duration;
 
 use common_meta::key::{CANDIDATES_ROOT, ELECTION_KEY};
-use common_telemetry::{error, warn};
+use common_telemetry::{error, info, warn};
 use common_time::Timestamp;
 use deadpool_postgres::{Manager, Pool};
 use snafu::{OptionExt, ResultExt, ensure};
@@ -477,6 +477,13 @@ impl Election for PgElection {
     }
 
     async fn reset_campaign(&self) {
+        info!("Resetting campaign");
+        if self.is_leader.load(Ordering::Relaxed) {
+            if let Err(err) = self.step_down_without_lock().await {
+                error!(err; "Failed to step down without lock");
+            }
+            info!("Step down without lock successfully, due to reset campaign");
+        }
         if let Err(err) = self.pg_client.write().await.reset_client().await {
             error!(err; "Failed to reset client");
         }
@@ -774,16 +781,12 @@ impl PgElection {
             key: key.clone(),
             ..Default::default()
         };
-        if self
-            .is_leader
-            .compare_exchange(true, false, Ordering::AcqRel, Ordering::Acquire)
-            .is_ok()
-            && let Err(e) = self
-                .leader_watcher
-                .send(LeaderChangeMessage::StepDown(Arc::new(leader_key)))
-        {
-            error!(e; "Failed to send leader change message");
-        }
+        send_leader_change_and_set_flags(
+            &self.is_leader,
+            &self.leader_infancy,
+            &self.leader_watcher,
+            LeaderChangeMessage::StepDown(Arc::new(leader_key)),
+        );
         Ok(())
     }
 
diff --git a/src/meta-srv/src/metasrv/builder.rs b/src/meta-srv/src/metasrv/builder.rs
@@ -373,7 +373,8 @@ impl MetasrvBuilder {
                 runtime_switch_manager.clone(),
                 meta_peer_client.clone(),
                 leader_cached_kv_backend.clone(),
-            );
+            )
+            .with_state(state.clone());
 
             Some(RegionFailureHandler::new(
                 region_supervisor,
diff --git a/src/meta-srv/src/region/supervisor.rs b/src/meta-srv/src/region/supervisor.rs
@@ -52,6 +52,7 @@ use crate::procedure::region_migration::{
 };
 use crate::region::failure_detector::RegionFailureDetector;
 use crate::selector::SelectorOptions;
+use crate::state::StateRef;
 
 /// `DatanodeHeartbeat` represents the heartbeat signal sent from a datanode.
 /// It includes identifiers for the cluster and datanode, a list of regions being monitored,
@@ -129,6 +130,9 @@ pub struct RegionSupervisorTicker {
     /// The [`Option`] wrapper allows us to abort the job while dropping the [`RegionSupervisor`].
     tick_handle: Mutex<Option<JoinHandle<()>>>,
 
+    /// The [`Option`] wrapper allows us to abort the job while dropping the [`RegionSupervisor`].
+    initialization_handler: Mutex<Option<JoinHandle<()>>>,
+
     /// The interval of tick.
     tick_interval: Duration,
 
@@ -172,6 +176,7 @@ impl RegionSupervisorTicker {
         );
         Self {
             tick_handle: Mutex::new(None),
+            initialization_handler: Mutex::new(None),
             tick_interval,
             initialization_delay,
             initialization_retry_period,
@@ -192,7 +197,7 @@ impl RegionSupervisorTicker {
                 self.initialization_retry_period,
             );
             initialization_interval.set_missed_tick_behavior(MissedTickBehavior::Skip);
-            common_runtime::spawn_global(async move {
+            let initialization_handler = common_runtime::spawn_global(async move {
                 loop {
                     initialization_interval.tick().await;
                     let (tx, rx) = oneshot::channel();
@@ -208,6 +213,7 @@ impl RegionSupervisorTicker {
                     }
                 }
             });
+            *self.initialization_handler.lock().unwrap() = Some(initialization_handler);
 
             let sender = self.sender.clone();
             let ticker_loop = tokio::spawn(async move {
@@ -237,6 +243,11 @@ impl RegionSupervisorTicker {
             handle.abort();
             info!("The tick loop is stopped.");
         }
+        let initialization_handler = self.initialization_handler.lock().unwrap().take();
+        if let Some(initialization_handler) = initialization_handler {
+            initialization_handler.abort();
+            info!("The initialization loop is stopped.");
+        }
     }
 }
 
@@ -280,6 +291,8 @@ pub struct RegionSupervisor {
     peer_resolver: PeerResolverRef,
     /// The kv backend.
     kv_backend: KvBackendRef,
+    /// The meta state, used to check if the current metasrv is the leader.
+    state: Option<StateRef>,
 }
 
 /// Controller for managing failure detectors for regions.
@@ -363,12 +376,29 @@ impl RegionSupervisor {
             runtime_switch_manager,
             peer_resolver,
             kv_backend,
+            state: None,
         }
     }
 
+    /// Sets the meta state.
+    pub(crate) fn with_state(mut self, state: StateRef) -> Self {
+        self.state = Some(state);
+        self
+    }
+
     /// Runs the main loop.
     pub(crate) async fn run(&mut self) {
         while let Some(event) = self.receiver.recv().await {
+            if let Some(state) = self.state.as_ref()
+                && !state.read().unwrap().is_leader()
+            {
+                warn!(
+                    "The current metasrv is not the leader, ignore {:?} event",
+                    event
+                );
+                continue;
+            }
+
             match event {
                 Event::InitializeAllRegions(sender) => {
                     match self.is_maintenance_mode_enabled().await {
@@ -403,7 +433,10 @@ impl RegionSupervisor {
                     self.deregister_failure_detectors(detecting_regions).await
                 }
                 Event::HeartbeatArrived(heartbeat) => self.on_heartbeat_arrived(heartbeat),
-                Event::Clear => self.clear(),
+                Event::Clear => {
+                    self.clear();
+                    info!("Region supervisor is initialized.");
+                }
                 #[cfg(test)]
                 Event::Dump(sender) => {
                     let _ = sender.send(self.failure_detector.dump());
@@ -896,6 +929,7 @@ pub(crate) mod tests {
         let (tx, mut rx) = tokio::sync::mpsc::channel(128);
         let ticker = RegionSupervisorTicker {
             tick_handle: Mutex::new(None),
+            initialization_handler: Mutex::new(None),
             tick_interval: Duration::from_millis(10),
             initialization_delay: Duration::from_millis(100),
             initialization_retry_period: Duration::from_millis(100),
@@ -922,6 +956,7 @@ pub(crate) mod tests {
         let (tx, mut rx) = tokio::sync::mpsc::channel(128);
         let ticker = RegionSupervisorTicker {
             tick_handle: Mutex::new(None),
+            initialization_handler: Mutex::new(None),
             tick_interval: Duration::from_millis(1000),
             initialization_delay: Duration::from_millis(50),
             initialization_retry_period: Duration::from_millis(50),
diff --git a/src/meta-srv/src/state.rs b/src/meta-srv/src/state.rs
@@ -75,6 +75,12 @@ impl State {
         })
     }
 
+    /// Returns true if the current state is a leader.
+    pub fn is_leader(&self) -> bool {
+        matches!(self, State::Leader(_))
+    }
+
+    /// Returns true if the leader cache is enabled.
     pub fn enable_leader_cache(&self) -> bool {
         match &self {
             State::Leader(leader) => leader.enable_leader_cache,
