feat: meta-client with new election mechanism (#7389)

**This section will be used as the commit message. Please do not leave this empty!**

This PR is a follow-up to PR #7179

This PR provides a meta client with an election client, and implements a sidecar rpc client update through the api of `members` and `leader` from LeaderService

But this also introduces a mutex, all meta client operations need to hold this lock

Approved-By: yezizp2012

ci/workflows/main-cron.yml

@@ -172,7 +172,7 @@ steps:
     retry: *auto-retry
 
   - label: "recovery test (deterministic simulation)"
-    command: "TEST_NUM=16 KILL_RATE=1.0 timeout 55m ci/scripts/deterministic-recovery-test.sh"
+    command: "TEST_NUM=12 KILL_RATE=1.0 timeout 55m ci/scripts/deterministic-recovery-test.sh"
     depends_on: "build-simulation"
     plugins:
       - gencer/cache#v2.4.10: *cargo-cache

ci/workflows/pull-request.yml

@@ -247,7 +247,7 @@ steps:
     retry: *auto-retry
 
   - label: "recovery test (deterministic simulation)"
-    command: "TEST_NUM=16 KILL_RATE=0.5 timeout 14m ci/scripts/deterministic-recovery-test.sh"
+    command: "TEST_NUM=8 KILL_RATE=0.5 timeout 14m ci/scripts/deterministic-recovery-test.sh"
     depends_on: "build-simulation"
     plugins:
       - gencer/cache#v2.4.10: *cargo-cache

src/prost/src/lib.rs

@@ -137,6 +137,9 @@ pub mod monitor_service_serde;
 #[rustfmt::skip]
 #[path = "backup_service.serde.rs"]
 pub mod backup_service_serde;
+#[rustfmt::skip]
+#[path = "leader.serde.rs"]
+pub mod leader_serde;
 
 #[derive(Clone, PartialEq, Eq, Debug)]
 pub struct ProstFieldNotFound(pub &'static str);

src/rpc_client/src/lib.rs

@@ -159,3 +159,21 @@ macro_rules! rpc_client_method_impl {
         )*
     }
 }
+
+#[macro_export]
+macro_rules! meta_rpc_client_method_impl {
+    ($( { $client:tt, $fn_name:ident, $req:ty, $resp:ty }),*) => {
+        $(
+            pub async fn $fn_name(&self, request: $req) -> $crate::Result<$resp> {
+                let mut client = self.core.read().await.$client.to_owned();
+                match client.$fn_name(request).await {
+                    Ok(resp) => Ok(resp.into_inner()),
+                    Err(e) => {
+                        self.refresh_client_if_needed(e.code()).await;
+                        Err(RpcError::GrpcStatus(e))
+                    }
+                }
+            }
+        )*
+    }
+}