fix: fetch leader_id from curp server directly

Refs: xline-kv#299

curp/src/client.rs

@@ -19,6 +19,8 @@ use crate::{
 
 /// Protocol client
 pub struct Client<C: Command> {
+    /// Self id
+    self_id: ServerId,
     /// Current leader and term
     state: RwLock<State>,
     /// All servers's `Connect`
@@ -91,8 +93,13 @@ where
 {
     /// Create a new protocol client based on the addresses
     #[inline]
-    pub async fn new(addrs: HashMap<ServerId, String>, timeout: ClientTimeout) -> Self {
+    pub async fn new(
+        self_id: ServerId,
+        addrs: HashMap<ServerId, String>,
+        timeout: ClientTimeout,
+    ) -> Self {
         Self {
+            self_id,
             state: RwLock::new(State::new()),
             connects: rpc::connect(addrs, None).await,
             timeout,
@@ -500,10 +507,27 @@ where
         }
     }
 
-    /// Get the current leader.
+    /// Fetch the current leader id and term from the curp server where is on the same node.
     #[inline]
-    pub fn leader(&self) -> Option<ServerId> {
-        self.state.read().leader.clone()
+    async fn fetch_local_leader_info(&self) -> Result<(Option<ServerId>, u64), ProposeError> {
+        let resp = self
+            .connects
+            .get(self.self_id.as_str())
+            .unwrap_or_else(|| unreachable!("self id {} not found", self.self_id.as_str()))
+            .fetch_leader(FetchLeaderRequest::new(), *self.timeout.retry_timeout())
+            .await?
+            .into_inner();
+
+        Ok((resp.leader_id, resp.term))
+    }
+
+    /// Fetch the current leader id without cache
+    #[inline]
+    pub async fn get_leader_id_from_curp(&self) -> ServerId {
+        if let Ok((Some(leader_id), _term)) = self.fetch_local_leader_info().await {
+            return leader_id;
+        }
+        self.fetch_leader().await
     }
 }
 

curp/tests/common/curp_group.rs

@@ -221,7 +221,7 @@ impl CurpGroup {
     }
 
     pub async fn new_client(&self, timeout: ClientTimeout) -> Client<TestCommand> {
-        Client::<TestCommand>::new(self.all.clone(), timeout).await
+        Client::<TestCommand>::new("S0".to_owned(), self.all.clone(), timeout).await
     }
 
     pub fn exe_rxs(

xline/src/client/mod.rs

@@ -68,11 +68,12 @@ impl Client {
         use_curp_client: bool,
         timeout: ClientTimeout,
     ) -> Result<Self, ClientError> {
+        let name = String::from("client");
         let etcd_client =
             EtcdClient::connect(all_members.values().cloned().collect_vec(), None).await?;
-        let curp_client = CurpClient::new(all_members, timeout).await;
+        let curp_client = CurpClient::new(name.clone(), all_members, timeout).await;
         Ok(Self {
-            name: String::from("client"),
+            name,
             curp_client,
             etcd_client,
             use_curp_client,

xline/src/server/lease_server.rs

@@ -259,7 +259,7 @@ where
             if self.lease_storage.is_primary() {
                 break self.leader_keep_alive(request_stream).await;
             }
-            let leader_id = self.client.get_leader_id().await;
+            let leader_id = self.client.get_leader_id_from_curp().await;
             // Given that a candidate server may become a leader when it won the election or
             // a follower when it lost the election. Therefore we need to double check here.
             // We can directly invoke leader_keep_alive when a candidate becomes a leader.

xline/src/server/xline_server.rs

@@ -277,7 +277,12 @@ impl XlineServer {
                 key_pair,
             )?;
         let client = Arc::new(
-            Client::<Command>::new(self.cluster_info.all_members(), self.client_timeout).await,
+            Client::<Command>::new(
+                self.cluster_info.self_id().clone(),
+                self.cluster_info.all_members(),
+                self.client_timeout,
+            )
+            .await,
         );
         let index_barrier = Arc::new(IndexBarrier::new());
         let id_barrier = Arc::new(IdBarrier::new());