refactor: support batch downgrade region instructions

Signed-off-by: WenyXu <wenymedia@gmail.com>

Author: WenyXu

src/common/meta/src/instruction.rs

@@ -55,6 +55,8 @@ impl Display for RegionIdent {
 /// The result of downgrade leader region.
 #[derive(Debug, Serialize, Deserialize, PartialEq, Eq, Clone)]
 pub struct DowngradeRegionReply {
+    #[serde(default)]
+    pub region_id: RegionId,
     /// Returns the `last_entry_id` if available.
     pub last_entry_id: Option<u64>,
     /// Returns the `metadata_last_entry_id` if available (Only available for metric engine).
@@ -423,8 +425,12 @@ pub enum Instruction {
     CloseRegions(Vec<RegionIdent>),
     /// Upgrades a region.
     UpgradeRegion(UpgradeRegion),
+    #[serde(
+        deserialize_with = "single_or_multiple_from",
+        alias = "DowngradeRegion"
+    )]
     /// Downgrades a region.
-    DowngradeRegion(DowngradeRegion),
+    DowngradeRegions(Vec<DowngradeRegion>),
     /// Invalidates batch cache.
     InvalidateCaches(Vec<CacheIdent>),
     /// Flushes regions.
@@ -457,9 +463,9 @@ impl Instruction {
     }
 
     /// Converts the instruction into a [DowngradeRegion].
-    pub fn into_downgrade_regions(self) -> Option<DowngradeRegion> {
+    pub fn into_downgrade_regions(self) -> Option<Vec<DowngradeRegion>> {
         match self {
-            Self::DowngradeRegion(downgrade_region) => Some(downgrade_region),
+            Self::DowngradeRegions(downgrade_region) => Some(downgrade_region),
             _ => None,
         }
     }
@@ -502,7 +508,11 @@ pub enum InstructionReply {
     #[serde(alias = "close_region")]
     CloseRegions(SimpleReply),
     UpgradeRegion(UpgradeRegionReply),
-    DowngradeRegion(DowngradeRegionReply),
+    #[serde(
+        alias = "downgrade_region",
+        deserialize_with = "single_or_multiple_from"
+    )]
+    DowngradeRegions(Vec<DowngradeRegionReply>),
     FlushRegions(FlushRegionReply),
 }
 
@@ -512,8 +522,8 @@ impl Display for InstructionReply {
             Self::OpenRegions(reply) => write!(f, "InstructionReply::OpenRegions({})", reply),
             Self::CloseRegions(reply) => write!(f, "InstructionReply::CloseRegions({})", reply),
             Self::UpgradeRegion(reply) => write!(f, "InstructionReply::UpgradeRegion({})", reply),
-            Self::DowngradeRegion(reply) => {
-                write!(f, "InstructionReply::DowngradeRegion({})", reply)
+            Self::DowngradeRegions(reply) => {
+                write!(f, "InstructionReply::DowngradeRegions({:?})", reply)
             }
             Self::FlushRegions(reply) => write!(f, "InstructionReply::FlushRegions({})", reply),
         }
@@ -543,9 +553,9 @@ impl InstructionReply {
         }
     }
 
-    pub fn expect_downgrade_region_reply(self) -> DowngradeRegionReply {
+    pub fn expect_downgrade_region_reply(self) -> Vec<DowngradeRegionReply> {
         match self {
-            Self::DowngradeRegion(reply) => reply,
+            Self::DowngradeRegions(reply) => reply,
             _ => panic!("Expected DowngradeRegion reply"),
         }
     }

src/datanode/src/heartbeat/handler.rs

@@ -100,7 +100,7 @@ impl RegionHeartbeatResponseHandler {
                 open_region_parallelism: self.open_region_parallelism,
             })),
             Instruction::FlushRegions(_) => Ok(Box::new(FlushRegionsHandler)),
-            Instruction::DowngradeRegion(_) => Ok(Box::new(DowngradeRegionsHandler)),
+            Instruction::DowngradeRegions(_) => Ok(Box::new(DowngradeRegionsHandler)),
             Instruction::UpgradeRegion(_) => Ok(Box::new(UpgradeRegionsHandler)),
             Instruction::InvalidateCaches(_) => InvalidHeartbeatResponseSnafu.fail(),
         }
@@ -112,7 +112,7 @@ impl HeartbeatResponseHandler for RegionHeartbeatResponseHandler {
     fn is_acceptable(&self, ctx: &HeartbeatResponseHandlerContext) -> bool {
         matches!(ctx.incoming_message.as_ref(), |Some((
             _,
-            Instruction::DowngradeRegion { .. },
+            Instruction::DowngradeRegions { .. },
         ))| Some((
             _,
             Instruction::UpgradeRegion { .. }
@@ -242,10 +242,10 @@ mod tests {
         );
 
         // Downgrade region
-        let instruction = Instruction::DowngradeRegion(DowngradeRegion {
+        let instruction = Instruction::DowngradeRegions(vec![DowngradeRegion {
             region_id: RegionId::new(2048, 1),
             flush_timeout: Some(Duration::from_secs(1)),
-        });
+        }]);
         assert!(
             heartbeat_handler
                 .is_acceptable(&heartbeat_env.create_handler_ctx((meta.clone(), instruction)))
@@ -440,44 +440,38 @@ mod tests {
         // Should be ok, if we try to downgrade it twice.
         for _ in 0..2 {
             let meta = MessageMeta::new_test(1, "test", "dn-1", "me-0");
-            let instruction = Instruction::DowngradeRegion(DowngradeRegion {
+            let instruction = Instruction::DowngradeRegions(vec![DowngradeRegion {
                 region_id,
                 flush_timeout: Some(Duration::from_secs(1)),
-            });
+            }]);
 
             let mut ctx = heartbeat_env.create_handler_ctx((meta, instruction));
             let control = heartbeat_handler.handle(&mut ctx).await.unwrap();
             assert_matches!(control, HandleControl::Continue);
 
             let (_, reply) = heartbeat_env.receiver.recv().await.unwrap();
 
-            if let InstructionReply::DowngradeRegion(reply) = reply {
-                assert!(reply.exists);
-                assert!(reply.error.is_none());
-                assert_eq!(reply.last_entry_id.unwrap(), 0);
-            } else {
-                unreachable!()
-            }
+            let reply = &reply.expect_downgrade_region_reply()[0];
+            assert!(reply.exists);
+            assert!(reply.error.is_none());
+            assert_eq!(reply.last_entry_id.unwrap(), 0);
         }
 
         // Downgrades a not exists region.
         let meta = MessageMeta::new_test(1, "test", "dn-1", "me-0");
-        let instruction = Instruction::DowngradeRegion(DowngradeRegion {
+        let instruction = Instruction::DowngradeRegions(vec![DowngradeRegion {
             region_id: RegionId::new(2048, 1),
             flush_timeout: Some(Duration::from_secs(1)),
-        });
+        }]);
         let mut ctx = heartbeat_env.create_handler_ctx((meta, instruction));
         let control = heartbeat_handler.handle(&mut ctx).await.unwrap();
         assert_matches!(control, HandleControl::Continue);
 
         let (_, reply) = heartbeat_env.receiver.recv().await.unwrap();
 
-        if let InstructionReply::DowngradeRegion(reply) = reply {
-            assert!(!reply.exists);
-            assert!(reply.error.is_none());
-            assert!(reply.last_entry_id.is_none());
-        } else {
-            unreachable!()
-        }
+        let reply = reply.expect_downgrade_region_reply();
+        assert!(!reply[0].exists);
+        assert!(reply[0].error.is_none());
+        assert!(reply[0].last_entry_id.is_none());
     }
 }