refactor: replace InactiveRegionManager with RegionLeaseKeeper

src/datanode/src/alive_keeper.rs

@@ -373,12 +373,16 @@ impl CountdownTask {
                             countdown.set(tokio::time::sleep_until(first_deadline));
                         },
                         Some(CountdownCommand::Reset((role, deadline))) => {
+                            // The first-time granted regions might be ignored because the `first_deadline` is larger than the `region_lease_timeout`.
+                            // Therefore, we set writable at the outside.
+                            // TODO(weny): Considers setting `first_deadline` to `region_lease_timeout`.
+                            let _ = self.region_server.set_writable(self.region_id, role.writable());
+
                             if countdown.deadline() < deadline {
                                 trace!(
                                     "Reset deadline of region {region_id} to approximately {} seconds later",
                                     (deadline - Instant::now()).as_secs_f32(),
                                 );
-                                let _ = self.region_server.set_writable(self.region_id, role.writable());
                                 countdown.set(tokio::time::sleep_until(deadline));
                             }
                             // Else the countdown could be either:

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

@@ -18,6 +18,7 @@ use api::v1::meta::HeartbeatRequest;
 use common_time::util as time_util;
 use serde::{Deserialize, Serialize};
 use store_api::region_engine::RegionRole;
+use store_api::storage::RegionId;
 
 use crate::error::{Error, InvalidHeartbeatRequestSnafu};
 use crate::keys::StatKey;
@@ -72,8 +73,12 @@ impl Stat {
         }
     }
 
-    pub fn region_ids(&self) -> Vec<u64> {
-        self.region_stats.iter().map(|s| s.id).collect()
+    /// Returns a tuple array containing [RegionId] and [RegionRole].
+    pub fn regions(&self) -> Vec<(RegionId, RegionRole)> {
+        self.region_stats
+            .iter()
+            .map(|s| (RegionId::from(s.id), s.role))
+            .collect()
     }
 
     pub fn retain_active_region_stats(&mut self, inactive_region_ids: &HashSet<u64>) {