feat: support to reject write after flushing

.github/workflows/develop.yml

@@ -442,6 +442,13 @@ jobs:
             minio: true
             kafka: true
             values: "with-remote-wal.yaml"
+        include:
+          - target: "fuzz_migrate_mito_regions"
+            mode:
+              name: "Local WAL"
+              minio: true
+              kafka: false
+              values: "with-minio.yaml"
     steps:
       - name: Remove unused software
         run: |
@@ -530,7 +537,7 @@ jobs:
         with:
           image-registry: localhost:5001
           values-filename: ${{ matrix.mode.values }}
-          enable-region-failover: true
+          enable-region-failover: ${{ matrix.mode.kafka }}
       - name: Port forward (mysql)
         run: |
           kubectl port-forward service/my-greptimedb-frontend 4002:4002 -n my-greptimedb&

Cargo.toml

@@ -120,7 +120,7 @@ etcd-client = { version = "0.13" }
 fst = "0.4.7"
 futures = "0.3"
 futures-util = "0.3"
-greptime-proto = { git = "https://github.com/GreptimeTeam/greptime-proto.git", rev = "36334744c7020734dcb4a6b8d24d52ae7ed53fe1" }
+greptime-proto = { git = "https://github.com/GreptimeTeam/greptime-proto.git", rev = "0b4f7c8ab06399f6b90e1626e8d5b9697cb33bb9" }
 humantime = "2.1"
 humantime-serde = "1.1"
 itertools = "0.10"

src/catalog/src/system_schema/information_schema/region_peers.rs

@@ -224,7 +224,7 @@ impl InformationSchemaRegionPeersBuilder {
             let region_id = RegionId::new(table_id, route.region.id.region_number()).as_u64();
             let peer_id = route.leader_peer.clone().map(|p| p.id);
             let peer_addr = route.leader_peer.clone().map(|p| p.addr);
-            let status = if let Some(status) = route.leader_status {
+            let status = if let Some(status) = route.leader_state {
                 Some(status.as_ref().to_string())
             } else {
                 // Alive by default

src/cmd/src/cli/bench.rs

@@ -158,7 +158,7 @@ fn create_region_routes(regions: Vec<RegionNumber>) -> Vec<RegionRoute> {
                 addr: String::new(),
             }),
             follower_peers: vec![],
-            leader_status: None,
+            leader_state: None,
             leader_down_since: None,
         });
     }

src/common/meta/src/ddl/alter_table/region_request.rs

@@ -187,7 +187,7 @@ mod tests {
                     region: Region::new_test(region_id),
                     leader_peer: Some(Peer::empty(1)),
                     follower_peers: vec![],
-                    leader_status: None,
+                    leader_state: None,
                     leader_down_since: None,
                 }]),
                 HashMap::new(),

src/common/meta/src/ddl/tests/alter_table.rs

@@ -107,21 +107,21 @@ async fn test_on_submit_alter_request() {
                     region: Region::new_test(RegionId::new(table_id, 1)),
                     leader_peer: Some(Peer::empty(1)),
                     follower_peers: vec![Peer::empty(5)],
-                    leader_status: None,
+                    leader_state: None,
                     leader_down_since: None,
                 },
                 RegionRoute {
                     region: Region::new_test(RegionId::new(table_id, 2)),
                     leader_peer: Some(Peer::empty(2)),
                     follower_peers: vec![Peer::empty(4)],
-                    leader_status: None,
+                    leader_state: None,
                     leader_down_since: None,
                 },
                 RegionRoute {
                     region: Region::new_test(RegionId::new(table_id, 3)),
                     leader_peer: Some(Peer::empty(3)),
                     follower_peers: vec![],
-                    leader_status: None,
+                    leader_state: None,
                     leader_down_since: None,
                 },
             ]),
@@ -193,21 +193,21 @@ async fn test_on_submit_alter_request_with_outdated_request() {
                     region: Region::new_test(RegionId::new(table_id, 1)),
                     leader_peer: Some(Peer::empty(1)),
                     follower_peers: vec![Peer::empty(5)],
-                    leader_status: None,
+                    leader_state: None,
                     leader_down_since: None,
                 },
                 RegionRoute {
                     region: Region::new_test(RegionId::new(table_id, 2)),
                     leader_peer: Some(Peer::empty(2)),
                     follower_peers: vec![Peer::empty(4)],
-                    leader_status: None,
+                    leader_state: None,
                     leader_down_since: None,
                 },
                 RegionRoute {
                     region: Region::new_test(RegionId::new(table_id, 3)),
                     leader_peer: Some(Peer::empty(3)),
                     follower_peers: vec![],
-                    leader_status: None,
+                    leader_state: None,
                     leader_down_since: None,
                 },
             ]),

src/common/meta/src/ddl/tests/drop_table.rs

@@ -119,21 +119,21 @@ async fn test_on_datanode_drop_regions() {
                     region: Region::new_test(RegionId::new(table_id, 1)),
                     leader_peer: Some(Peer::empty(1)),
                     follower_peers: vec![Peer::empty(5)],
-                    leader_status: None,
+                    leader_state: None,
                     leader_down_since: None,
                 },
                 RegionRoute {
                     region: Region::new_test(RegionId::new(table_id, 2)),
                     leader_peer: Some(Peer::empty(2)),
                     follower_peers: vec![Peer::empty(4)],
-                    leader_status: None,
+                    leader_state: None,
                     leader_down_since: None,
                 },
                 RegionRoute {
                     region: Region::new_test(RegionId::new(table_id, 3)),
                     leader_peer: Some(Peer::empty(3)),
                     follower_peers: vec![],
-                    leader_status: None,
+                    leader_state: None,
                     leader_down_since: None,
                 },
             ]),

src/common/meta/src/instruction.rs

@@ -137,14 +137,16 @@ pub struct DowngradeRegion {
     /// `None` stands for don't flush before downgrading the region.
     #[serde(default)]
     pub flush_timeout: Option<Duration>,
+    /// Rejects all write requests after flushing.
+    pub reject_write: bool,
 }
 
 impl Display for DowngradeRegion {
     fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
         write!(
             f,
-            "DowngradeRegion(region_id={}, flush_timeout={:?})",
-            self.region_id, self.flush_timeout,
+            "DowngradeRegion(region_id={}, flush_timeout={:?}, rejct_write={})",
+            self.region_id, self.flush_timeout, self.reject_write
         )
     }
 }

src/common/meta/src/key.rs

@@ -140,7 +140,7 @@ use crate::key::table_route::TableRouteKey;
 use crate::key::txn_helper::TxnOpGetResponseSet;
 use crate::kv_backend::txn::{Txn, TxnOp};
 use crate::kv_backend::KvBackendRef;
-use crate::rpc::router::{region_distribution, RegionRoute, RegionStatus};
+use crate::rpc::router::{region_distribution, LeaderState, RegionRoute};
 use crate::rpc::store::BatchDeleteRequest;
 use crate::DatanodeId;
 
@@ -1126,14 +1126,14 @@ impl TableMetadataManager {
         next_region_route_status: F,
     ) -> Result<()>
     where
-        F: Fn(&RegionRoute) -> Option<Option<RegionStatus>>,
+        F: Fn(&RegionRoute) -> Option<Option<LeaderState>>,
     {
         let mut new_region_routes = current_table_route_value.region_routes()?.clone();
 
         let mut updated = 0;
         for route in &mut new_region_routes {
             if let Some(status) = next_region_route_status(route) {
-                if route.set_leader_status(status) {
+                if route.set_leader_state(status) {
                     updated += 1;
                 }
             }
@@ -1280,7 +1280,7 @@ mod tests {
     use crate::key::{DeserializedValueWithBytes, TableMetadataManager, ViewInfoValue};
     use crate::kv_backend::memory::MemoryKvBackend;
     use crate::peer::Peer;
-    use crate::rpc::router::{region_distribution, Region, RegionRoute, RegionStatus};
+    use crate::rpc::router::{region_distribution, LeaderState, Region, RegionRoute};
 
     #[test]
     fn test_deserialized_value_with_bytes() {
@@ -1324,7 +1324,7 @@ mod tests {
             },
             leader_peer: Some(Peer::new(datanode, "a2")),
             follower_peers: vec![],
-            leader_status: None,
+            leader_state: None,
             leader_down_since: None,
         }
     }
@@ -1715,7 +1715,7 @@ mod tests {
                     attrs: BTreeMap::new(),
                 },
                 leader_peer: Some(Peer::new(datanode, "a2")),
-                leader_status: Some(RegionStatus::Downgraded),
+                leader_state: Some(LeaderState::Downgrading),
                 follower_peers: vec![],
                 leader_down_since: Some(current_time_millis()),
             },
@@ -1727,7 +1727,7 @@ mod tests {
                     attrs: BTreeMap::new(),
                 },
                 leader_peer: Some(Peer::new(datanode, "a1")),
-                leader_status: None,
+                leader_state: None,
                 follower_peers: vec![],
                 leader_down_since: None,
             },
@@ -1750,10 +1750,10 @@ mod tests {
 
         table_metadata_manager
             .update_leader_region_status(table_id, &current_table_route_value, |region_route| {
-                if region_route.leader_status.is_some() {
+                if region_route.leader_state.is_some() {
                     None
                 } else {
-                    Some(Some(RegionStatus::Downgraded))
+                    Some(Some(LeaderState::Downgrading))
                 }
             })
             .await
@@ -1768,17 +1768,17 @@ mod tests {
             .unwrap();
 
         assert_eq!(
-            updated_route_value.region_routes().unwrap()[0].leader_status,
-            Some(RegionStatus::Downgraded)
+            updated_route_value.region_routes().unwrap()[0].leader_state,
+            Some(LeaderState::Downgrading)
         );
 
         assert!(updated_route_value.region_routes().unwrap()[0]
             .leader_down_since
             .is_some());
 
         assert_eq!(
-            updated_route_value.region_routes().unwrap()[1].leader_status,
-            Some(RegionStatus::Downgraded)
+            updated_route_value.region_routes().unwrap()[1].leader_state,
+            Some(LeaderState::Downgrading)
         );
         assert!(updated_route_value.region_routes().unwrap()[1]
             .leader_down_since
@@ -1943,21 +1943,21 @@ mod tests {
                         region: Region::new_test(RegionId::new(table_id, 1)),
                         leader_peer: Some(Peer::empty(1)),
                         follower_peers: vec![Peer::empty(5)],
-                        leader_status: None,
+                        leader_state: None,
                         leader_down_since: None,
                     },
                     RegionRoute {
                         region: Region::new_test(RegionId::new(table_id, 2)),
                         leader_peer: Some(Peer::empty(2)),
                         follower_peers: vec![Peer::empty(4)],
-                        leader_status: None,
+                        leader_state: None,
                         leader_down_since: None,
                     },
                     RegionRoute {
                         region: Region::new_test(RegionId::new(table_id, 3)),
                         leader_peer: Some(Peer::empty(3)),
                         follower_peers: vec![],
-                        leader_status: None,
+                        leader_state: None,
                         leader_down_since: None,
                     },
                 ]),
@@ -1996,21 +1996,21 @@ mod tests {
                         region: Region::new_test(RegionId::new(table_id, 1)),
                         leader_peer: Some(Peer::empty(1)),
                         follower_peers: vec![Peer::empty(5)],
-                        leader_status: None,
+                        leader_state: None,
                         leader_down_since: None,
                     },
                     RegionRoute {
                         region: Region::new_test(RegionId::new(table_id, 2)),
                         leader_peer: Some(Peer::empty(2)),
                         follower_peers: vec![Peer::empty(4)],
-                        leader_status: None,
+                        leader_state: None,
                         leader_down_since: None,
                     },
                     RegionRoute {
                         region: Region::new_test(RegionId::new(table_id, 3)),
                         leader_peer: Some(Peer::empty(3)),
                         follower_peers: vec![],
-                        leader_status: None,
+                        leader_state: None,
                         leader_down_since: None,
                     },
                 ]),

src/common/meta/src/key/table_route.rs

@@ -744,18 +744,51 @@ mod tests {
     use crate::kv_backend::memory::MemoryKvBackend;
     use crate::kv_backend::{KvBackend, TxnService};
     use crate::peer::Peer;
+    use crate::rpc::router::Region;
     use crate::rpc::store::PutRequest;
 
     #[test]
     fn test_table_route_compatibility() {
         let old_raw_v = r#"{"region_routes":[{"region":{"id":1,"name":"r1","partition":null,"attrs":{}},"leader_peer":{"id":2,"addr":"a2"},"follower_peers":[]},{"region":{"id":1,"name":"r1","partition":null,"attrs":{}},"leader_peer":{"id":2,"addr":"a2"},"follower_peers":[]}],"version":0}"#;
         let v = TableRouteValue::try_from_raw_value(old_raw_v.as_bytes()).unwrap();
 
-        let new_raw_v = format!("{:?}", v);
-        assert_eq!(
-            new_raw_v,
-            r#"Physical(PhysicalTableRouteValue { region_routes: [RegionRoute { region: Region { id: 1(0, 1), name: "r1", partition: None, attrs: {} }, leader_peer: Some(Peer { id: 2, addr: "a2" }), follower_peers: [], leader_status: None, leader_down_since: None }, RegionRoute { region: Region { id: 1(0, 1), name: "r1", partition: None, attrs: {} }, leader_peer: Some(Peer { id: 2, addr: "a2" }), follower_peers: [], leader_status: None, leader_down_since: None }], version: 0 })"#
-        );
+        let expected_table_route = TableRouteValue::Physical(PhysicalTableRouteValue {
+            region_routes: vec![
+                RegionRoute {
+                    region: Region {
+                        id: RegionId::new(0, 1),
+                        name: "r1".to_string(),
+                        partition: None,
+                        attrs: Default::default(),
+                    },
+                    leader_peer: Some(Peer {
+                        id: 2,
+                        addr: "a2".to_string(),
+                    }),
+                    follower_peers: vec![],
+                    leader_state: None,
+                    leader_down_since: None,
+                },
+                RegionRoute {
+                    region: Region {
+                        id: RegionId::new(0, 1),
+                        name: "r1".to_string(),
+                        partition: None,
+                        attrs: Default::default(),
+                    },
+                    leader_peer: Some(Peer {
+                        id: 2,
+                        addr: "a2".to_string(),
+                    }),
+                    follower_peers: vec![],
+                    leader_state: None,
+                    leader_down_since: None,
+                },
+            ],
+            version: 0,
+        });
+
+        assert_eq!(v, expected_table_route);
     }
 
     #[test]

src/common/meta/src/region_keeper.rs

@@ -58,7 +58,7 @@ impl MemoryRegionKeeper {
         Default::default()
     }
 
-    /// Returns [OpeningRegionGuard] if Region(`region_id`) on Peer(`datanode_id`) does not exist.
+    /// Returns [OperatingRegionGuard] if Region(`region_id`) on Peer(`datanode_id`) does not exist.
     pub fn register(
         &self,
         datanode_id: DatanodeId,