refactor: extract leader_or_forward() from gRPC handler functions

Three handlers (`handle_kv_transaction`, `handle_kv_list`, `handle_kv_get_many`)
duplicated the same 6-line "assume leader or return forward-to-leader Status"
block. Extract it into `leader_or_forward()` so each handler reduces to a
single `self.leader_or_forward().await?` call.

The post-write forward check in `handle_kv_transaction` is intentionally kept
inline — it handles a different error path (`ClientWriteError::ForwardToLeader`)
for leadership changes that occur between the assume check and the raft write.

Author: Zhang Yanpo

crates/common/types/src/cmd/mod.rs

@@ -179,7 +179,7 @@ mod tests {
         let want = concat!(
             r#"{"Transaction":{"#,
             r#""condition":[{"key":"k","expected":0,"target":{"Value":[118]}}],"#,
-            r#""if_then":[{"request":{"Put":{"key":"k","value":[118],"expire_at":null,"ttl_ms":100}}}],"#,
+            r#""if_then":[{"request":{"Put":{"key":"k","value":[118],"expire_at":null,"ttl_ms":100,"match_seq":null}}}],"#,
             r#""else_then":[]"#,
             r#"}}"#
         );

crates/common/types/tests/it/txn_serde.rs

@@ -28,7 +28,7 @@ fn test_txn_request_serde() -> anyhow::Result<()> {
     let want = concat!(
         r#"{"#,
         r#""condition":[{"key":"k","expected":0,"target":{"Value":[118]}}],"#,
-        r#""if_then":[{"request":{"Put":{"key":"k","value":[118],"expire_at":null,"ttl_ms":100}}}],"#,
+        r#""if_then":[{"request":{"Put":{"key":"k","value":[118],"expire_at":null,"ttl_ms":100,"match_seq":null}}}],"#,
         r#""else_then":[]"#,
         r#"}"#
     );

crates/server/service/src/meta_node/meta_node.rs

@@ -1639,20 +1639,27 @@ impl<SP: SpawnApi> MetaNode<SP> {
         Endpoint::parse(addr).ok()
     }
 
+    /// Return a `MetaLeader` if this node is the leader.
+    ///
+    /// Otherwise, resolve the leader endpoint and return a forward-to-leader gRPC `Status`.
+    async fn leader_or_forward(&self) -> Result<MetaLeader<'_, SP>, Status> {
+        match self.assume_leader().await {
+            Ok(leader) => Ok(leader),
+            Err(forward) => {
+                let endpoint = self.get_leader_endpoint(forward.leader_id).await;
+                Err(GrpcHelper::status_forward_to_leader(endpoint.as_ref()))
+            }
+        }
+    }
+
     /// Handle KvTransaction request. Must be leader to process.
     ///
     /// If this node is not the leader, returns a `Status` error with leader endpoint in metadata.
     pub async fn handle_kv_transaction(
         &self,
         txn: kv_transaction::Transaction,
     ) -> Result<KvTransactionReply, Status> {
-        let leader = match self.assume_leader().await {
-            Ok(leader) => leader,
-            Err(forward) => {
-                let endpoint = self.get_leader_endpoint(forward.leader_id).await;
-                return Err(GrpcHelper::status_forward_to_leader(endpoint.as_ref()));
-            }
-        };
+        let leader = self.leader_or_forward().await?;
 
         let entry = LogEntry::new(Cmd::KvTransaction(txn));
         let applied = match leader.write(entry).await {
@@ -1678,13 +1685,7 @@ impl<SP: SpawnApi> MetaNode<SP> {
         prefix: String,
         limit: Option<u64>,
     ) -> Result<BoxStream<'static, Result<StreamItem, Status>>, Status> {
-        let leader = match self.assume_leader().await {
-            Ok(leader) => leader,
-            Err(forward) => {
-                let endpoint = self.get_leader_endpoint(forward.leader_id).await;
-                return Err(GrpcHelper::status_forward_to_leader(endpoint.as_ref()));
-            }
-        };
+        let leader = self.leader_or_forward().await?;
 
         let strm = leader
             .kv_list(&prefix, limit)
@@ -1702,13 +1703,7 @@ impl<SP: SpawnApi> MetaNode<SP> {
         &self,
         input: impl Stream<Item = Result<KvGetManyRequest, Status>> + Send + 'static,
     ) -> Result<BoxStream<'static, Result<StreamItem, Status>>, Status> {
-        let leader = match self.assume_leader().await {
-            Ok(leader) => leader,
-            Err(forward) => {
-                let endpoint = self.get_leader_endpoint(forward.leader_id).await;
-                return Err(GrpcHelper::status_forward_to_leader(endpoint.as_ref()));
-            }
-        };
+        let leader = self.leader_or_forward().await?;
 
         let strm = leader
             .kv_get_many(input)

crates/server/service/tests/it/grpc/metasrv_grpc_export.rs

@@ -87,9 +87,9 @@ async fn test_export() -> anyhow::Result<()> {
         r#"["raft_log",{"LogEntry":{"log_id":{"leader_id":{"term":1,"node_id":0},"index":1},"payload":"Blank"}}]"#,
         r#"["raft_log",{"LogEntry":{"log_id":{"leader_id":{"term":1,"node_id":0},"index":2},"payload":{"Normal":{"time_ms":1111111111111,"cmd":{"AddNode":{"node_id":0,"node":{"name":"0","endpoint":{"addr":"localhost","port":29000},"grpc_api_advertise_address":"127.0.0.1:29000"},"overriding":false}}}}}}]"#,
         r#"["raft_log",{"LogEntry":{"log_id":{"leader_id":{"term":1,"node_id":0},"index":3},"payload":{"Membership":{"configs":[[0]],"nodes":{"0":{}}}}}}]"#,
-        r#"["raft_log",{"LogEntry":{"log_id":{"leader_id":{"term":1,"node_id":0},"index":4},"payload":{"Normal":{"time_ms":1111111111111,"cmd":{"Transaction":{"condition":[{"key":"foo","expected":2,"target":{"Seq":0}}],"if_then":[{"request":{"Put":{"key":"foo","value":[102,111,111],"expire_at":null}}}],"else_then":[{"request":{"Get":{"key":"foo"}}}]}}}}}}]"#,
-        r#"["raft_log",{"LogEntry":{"log_id":{"leader_id":{"term":1,"node_id":0},"index":5},"payload":{"Normal":{"time_ms":1111111111111,"cmd":{"Transaction":{"condition":[{"key":"bar","expected":2,"target":{"Seq":0}}],"if_then":[{"request":{"Put":{"key":"bar","value":[98,97,114],"expire_at":null}}}],"else_then":[{"request":{"Get":{"key":"bar"}}}]}}}}}}]"#,
-        r#"["raft_log",{"LogEntry":{"log_id":{"leader_id":{"term":1,"node_id":0},"index":6},"payload":{"Normal":{"time_ms":1111111111111,"cmd":{"Transaction":{"condition":[{"key":"wow","expected":2,"target":{"Seq":0}}],"if_then":[{"request":{"Put":{"key":"wow","value":[119,111,119],"expire_at":null}}}],"else_then":[{"request":{"Get":{"key":"wow"}}}]}}}}}}]"#,
+        r#"["raft_log",{"LogEntry":{"log_id":{"leader_id":{"term":1,"node_id":0},"index":4},"payload":{"Normal":{"time_ms":1111111111111,"cmd":{"Transaction":{"condition":[{"key":"foo","expected":2,"target":{"Seq":0}}],"if_then":[{"request":{"Put":{"key":"foo","value":[102,111,111],"expire_at":null,"match_seq":null}}}],"else_then":[{"request":{"Get":{"key":"foo"}}}]}}}}}}]"#,
+        r#"["raft_log",{"LogEntry":{"log_id":{"leader_id":{"term":1,"node_id":0},"index":5},"payload":{"Normal":{"time_ms":1111111111111,"cmd":{"Transaction":{"condition":[{"key":"bar","expected":2,"target":{"Seq":0}}],"if_then":[{"request":{"Put":{"key":"bar","value":[98,97,114],"expire_at":null,"match_seq":null}}}],"else_then":[{"request":{"Get":{"key":"bar"}}}]}}}}}}]"#,
+        r#"["raft_log",{"LogEntry":{"log_id":{"leader_id":{"term":1,"node_id":0},"index":6},"payload":{"Normal":{"time_ms":1111111111111,"cmd":{"Transaction":{"condition":[{"key":"wow","expected":2,"target":{"Seq":0}}],"if_then":[{"request":{"Put":{"key":"wow","value":[119,111,119],"expire_at":null,"match_seq":null}}}],"else_then":[{"request":{"Get":{"key":"wow"}}}]}}}}}}]"#,
         r#"["state_machine/0",{"Sequences":{"key":"generic-kv","value":3}}]"#,
         r#"["state_machine/0",{"StateMachineMeta":{"key":"LastApplied","value":{"LogId":{"leader_id":{"term":1,"node_id":0},"index":6}}}}]"#,
         r#"["state_machine/0",{"StateMachineMeta":{"key":"LastMembership","value":{"Membership":{"log_id":{"leader_id":{"term":1,"node_id":0},"index":3},"membership":{"configs":[[0]],"nodes":{"0":{}}}}}}}]"#,