refactor: migrate region failover implementation to region migration

src/common/meta/src/wal_options_allocator.rs

@@ -87,6 +87,11 @@ impl WalOptionsAllocator {
             }
         }
     }
+
+    /// Returns true if it's the remote WAL.
+    pub fn is_remote_wal(&self) -> bool {
+        matches!(&self, WalOptionsAllocator::Kafka(_))
+    }
 }
 
 /// Allocates a wal options for each region. The allocated wal options is encoded immediately.

src/meta-srv/src/error.rs

@@ -840,6 +840,13 @@ pub enum Error {
         #[snafu(source(from(common_config::error::Error, Box::new)))]
         source: Box<common_config::error::Error>,
     },
+
+    #[snafu(display("Region: {} leader peer is not found", region_id))]
+    RegionLeaderNotFound {
+        #[snafu(implicit)]
+        location: Location,
+        region_id: RegionId,
+    },
 }
 
 impl Error {
@@ -931,7 +938,8 @@ impl ErrorExt for Error {
             | Error::RegionOpeningRace { .. }
             | Error::RegionRouteNotFound { .. }
             | Error::MigrationAbort { .. }
-            | Error::MigrationRunning { .. } => StatusCode::Unexpected,
+            | Error::MigrationRunning { .. }
+            | Error::RegionLeaderNotFound { .. } => StatusCode::Unexpected,
             Error::TableNotFound { .. } => StatusCode::TableNotFound,
             Error::SaveClusterInfo { source, .. }
             | Error::InvalidClusterInfoFormat { source, .. } => source.status_code(),

src/meta-srv/src/failure_detector.rs

@@ -63,7 +63,7 @@ pub(crate) struct PhiAccrualFailureDetector {
     last_heartbeat_millis: Option<i64>,
 }
 
-#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
+#[derive(Clone, Copy, Debug, PartialEq, Serialize, Deserialize)]
 #[serde(default)]
 pub struct PhiAccrualFailureDetectorOptions {
     pub threshold: f32,
@@ -195,7 +195,7 @@ fn phi(time_diff: i64, mean: f64, std_deviation: f64) -> f64 {
 /// It is capped by the number of samples specified in `max_sample_size`.
 ///
 /// The stats (mean, variance, std_deviation) are not defined for empty HeartbeatHistory.
-#[derive(Clone)]
+#[derive(Debug, Clone)]
 struct HeartbeatHistory {
     /// Number of samples to use for calculation of mean and standard deviation of inter-arrival
     /// times.

src/meta-srv/src/handler/failure_handler.rs

@@ -12,48 +12,25 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-mod runner;
-
-use std::sync::Arc;
-
 use api::v1::meta::{HeartbeatRequest, Role};
 use async_trait::async_trait;
-use common_meta::RegionIdent;
+use common_telemetry::info;
 
 use crate::error::Result;
-use crate::failure_detector::PhiAccrualFailureDetectorOptions;
-use crate::handler::failure_handler::runner::{FailureDetectControl, FailureDetectRunner};
 use crate::handler::{HandleControl, HeartbeatAccumulator, HeartbeatHandler};
-use crate::metasrv::{Context, ElectionRef};
-use crate::procedure::region_failover::RegionFailoverManager;
-
-pub(crate) struct DatanodeHeartbeat {
-    region_idents: Vec<RegionIdent>,
-    heartbeat_time: i64,
-}
+use crate::metasrv::Context;
+use crate::region::supervisor::{DatanodeHeartbeat, HeartbeatAcceptor, RegionSupervisor};
 
 pub struct RegionFailureHandler {
-    failure_detect_runner: FailureDetectRunner,
+    heartbeat_acceptor: HeartbeatAcceptor,
 }
 
 impl RegionFailureHandler {
-    pub(crate) async fn try_new(
-        election: Option<ElectionRef>,
-        region_failover_manager: Arc<RegionFailoverManager>,
-        failure_detector_options: PhiAccrualFailureDetectorOptions,
-    ) -> Result<Self> {
-        region_failover_manager.try_start()?;
-
-        let mut failure_detect_runner = FailureDetectRunner::new(
-            election,
-            region_failover_manager.clone(),
-            failure_detector_options,
-        );
-        failure_detect_runner.start().await;
-
-        Ok(Self {
-            failure_detect_runner,
-        })
+    pub(crate) fn new(mut region_supervisor: RegionSupervisor) -> Self {
+        let heartbeat_acceptor = region_supervisor.heartbeat_acceptor();
+        info!("Starting region supervisor");
+        tokio::spawn(async move { region_supervisor.run().await });
+        Self { heartbeat_acceptor }
     }
 }
 
@@ -66,73 +43,45 @@ impl HeartbeatHandler for RegionFailureHandler {
     async fn handle(
         &self,
         _: &HeartbeatRequest,
-        ctx: &mut Context,
+        _ctx: &mut Context,
         acc: &mut HeartbeatAccumulator,
     ) -> Result<HandleControl> {
-        if ctx.is_infancy {
-            self.failure_detect_runner
-                .send_control(FailureDetectControl::Purge)
-                .await;
-        }
-
         let Some(stat) = acc.stat.as_ref() else {
             return Ok(HandleControl::Continue);
         };
 
-        let heartbeat = DatanodeHeartbeat {
-            region_idents: stat
-                .region_stats
-                .iter()
-                .map(|x| {
-                    let region_id = x.id;
-                    RegionIdent {
-                        cluster_id: stat.cluster_id,
-                        datanode_id: stat.id,
-                        table_id: region_id.table_id(),
-                        region_number: region_id.region_number(),
-                        engine: x.engine.clone(),
-                    }
-                })
-                .collect(),
-            heartbeat_time: stat.timestamp_millis,
-        };
-
-        self.failure_detect_runner.send_heartbeat(heartbeat).await;
+        self.heartbeat_acceptor
+            .accept(DatanodeHeartbeat::from(stat))
+            .await;
 
         Ok(HandleControl::Continue)
     }
 }
 
 #[cfg(test)]
 mod tests {
-    use std::assert_matches::assert_matches;
-
+    use api::v1::meta::HeartbeatRequest;
     use common_catalog::consts::default_engine;
-    use common_meta::key::MAINTENANCE_KEY;
     use store_api::region_engine::RegionRole;
     use store_api::storage::RegionId;
+    use tokio::sync::oneshot;
 
-    use super::*;
+    use crate::handler::failure_handler::RegionFailureHandler;
     use crate::handler::node_stat::{RegionStat, Stat};
+    use crate::handler::{HeartbeatAccumulator, HeartbeatHandler};
     use crate::metasrv::builder::MetasrvBuilder;
-    use crate::test_util::create_region_failover_manager;
+    use crate::region::supervisor::tests::new_test_supervisor;
+    use crate::region::supervisor::Event;
 
-    #[tokio::test(flavor = "multi_thread")]
+    #[tokio::test]
     async fn test_handle_heartbeat() {
-        let region_failover_manager = create_region_failover_manager();
-        let failure_detector_options = PhiAccrualFailureDetectorOptions::default();
-        let handler =
-            RegionFailureHandler::try_new(None, region_failover_manager, failure_detector_options)
-                .await
-                .unwrap();
-
+        let supervisor = new_test_supervisor();
+        let sender = supervisor.sender();
+        let handler = RegionFailureHandler::new(supervisor);
         let req = &HeartbeatRequest::default();
-
         let builder = MetasrvBuilder::new();
         let metasrv = builder.build().await.unwrap();
         let mut ctx = metasrv.new_ctx();
-        ctx.is_infancy = false;
-
         let acc = &mut HeartbeatAccumulator::default();
         fn new_region_stat(region_id: u64) -> RegionStat {
             RegionStat {
@@ -153,48 +102,9 @@ mod tests {
         });
 
         handler.handle(req, &mut ctx, acc).await.unwrap();
-
-        let dump = handler.failure_detect_runner.dump().await;
-        assert_eq!(dump.iter().collect::<Vec<_>>().len(), 3);
-
-        // infancy makes heartbeats re-accumulated
-        ctx.is_infancy = true;
-        acc.stat = None;
-        handler.handle(req, &mut ctx, acc).await.unwrap();
-        let dump = handler.failure_detect_runner.dump().await;
-        assert_eq!(dump.iter().collect::<Vec<_>>().len(), 0);
-    }
-
-    #[tokio::test(flavor = "multi_thread")]
-    async fn test_maintenance_mode() {
-        let region_failover_manager = create_region_failover_manager();
-        let kv_backend = region_failover_manager.create_context().kv_backend.clone();
-        let _handler = RegionFailureHandler::try_new(
-            None,
-            region_failover_manager.clone(),
-            PhiAccrualFailureDetectorOptions::default(),
-        )
-        .await
-        .unwrap();
-
-        let kv_req = common_meta::rpc::store::PutRequest {
-            key: Vec::from(MAINTENANCE_KEY),
-            value: vec![],
-            prev_kv: false,
-        };
-        let _ = kv_backend.put(kv_req.clone()).await.unwrap();
-        assert_matches!(
-            region_failover_manager.is_maintenance_mode().await,
-            Ok(true)
-        );
-
-        let _ = kv_backend
-            .delete(MAINTENANCE_KEY.as_bytes(), false)
-            .await
-            .unwrap();
-        assert_matches!(
-            region_failover_manager.is_maintenance_mode().await,
-            Ok(false)
-        );
+        let (tx, rx) = oneshot::channel();
+        sender.send(Event::Dump(tx)).await.unwrap();
+        let detector = rx.await.unwrap();
+        assert_eq!(detector.iter().collect::<Vec<_>>().len(), 3);
     }
 }