refactor: Remove install_snapshot v0 meta-service API (#15590)

**Compatibility Changes:**

This commit introduces changes to inter-meta-service compatibility,
although compatibility between databend-query and meta-service
remains unaffected.

**Details:**

- The initial snapshot transmission API, `install_snapshot` (v0), has
  been removed from both the meta server and meta client sides.

- For clients, the use of `install_snapshot_v1()` transitions from
  **OPTIONAL** to **REQUIRED**.

**Version Requirements:**

- Post this commit, the meta-server requires clients to be at least
  version `1.2.212`, starting from which the clients began using
  `install_snapshot_v1()`.

- Similarly, clients require the server to be at least version
  `1.2.212`, which is the version where `install_snapshot_v1()` becomes
  available.

Author: drmingdrmer

src/meta/README.md

@@ -96,10 +96,13 @@ History versions that are not included in the above chart:
 
 ## Compatibility between databend-meta
 
-| Meta version      | Backward compatible with |
-|:------------------|:-------------------------|
-| [0.9.41, 1.2.212) | [0.9.41, 1.2.212)        |
-| [1.2.212, +∞)     | [0.9.41, +∞)             |
+| Meta version         | Backward compatible with |
+|:---------------------|:-------------------------|
+| [0.9.41,   1.2.212)  | [0.9.41,  1.2.212)       |
+| [1.2.212,  1.2.476?) | [0.9.41,  1.2.476?)      |
+| [1.2.476?, +∞)       | [1.2.212, +∞)            |
+
+TODO: fix the above version when merged
 
 
 - `1.2.53` Incompatible, rolling upgrade is allowed without snapshot transmitting.
@@ -113,6 +116,9 @@ History versions that are not included in the above chart:
   In this version, databend-meta raft-server introduced a new API `install_snapshot_v1()`.
   The raft-client will try to use either this new API or the original `install_snapshot()`.
 
+- `1.2.476?` Remove: `install_snapshot()`(v0) from client and server.
+  The `install_snapshot_v1()` is the only API to install snapshot, and becomes **REQUIRED** for the client.
+    
 
 ## Compatibility of databend-meta on-disk data
 

src/meta/service/src/meta_service/raft_service_impl.rs

@@ -75,27 +75,6 @@ impl RaftServiceImpl {
         }
     }
 
-    async fn do_install_snapshot(
-        &self,
-        request: Request<RaftRequest>,
-    ) -> Result<Response<RaftReply>, Status> {
-        let addr = remote_addr(&request);
-
-        self.incr_meta_metrics_recv_bytes_from_peer(&request);
-        let _g = snapshot_recv_inflight(&addr).counter_guard();
-
-        let is_req = GrpcHelper::parse_req(request)?;
-
-        let res = self
-            .receive_chunked_snapshot(is_req)
-            .timed(observe_snapshot_recv_spent(&addr))
-            .await;
-
-        raft_metrics::network::incr_snapshot_recvfrom_result(addr.clone(), res.is_ok());
-
-        GrpcHelper::make_grpc_result(res)
-    }
-
     async fn do_install_snapshot_v1(
         &self,
         request: Request<SnapshotChunkRequest>,
@@ -273,14 +252,6 @@ impl RaftService for RaftServiceImpl {
         .await
     }
 
-    async fn install_snapshot(
-        &self,
-        request: Request<RaftRequest>,
-    ) -> Result<Response<RaftReply>, Status> {
-        let root = databend_common_tracing::start_trace_for_remote_request(full_name!(), &request);
-        self.do_install_snapshot(request).in_span(root).await
-    }
-
     async fn install_snapshot_v1(
         &self,
         request: Request<SnapshotChunkRequest>,

src/meta/service/src/network.rs

@@ -33,6 +33,7 @@ use databend_common_meta_sled_store::openraft::error::Unreachable;
 use databend_common_meta_sled_store::openraft::network::RPCOption;
 use databend_common_meta_sled_store::openraft::MessageSummary;
 use databend_common_meta_sled_store::openraft::RaftNetworkFactory;
+use databend_common_meta_types::protobuf::RaftReply;
 use databend_common_meta_types::protobuf::RaftRequest;
 use databend_common_meta_types::protobuf::SnapshotChunkRequest;
 use databend_common_meta_types::AppendEntriesRequest;
@@ -173,15 +174,12 @@ pub struct NetworkConnection {
     target: NodeId,
 
     /// The node info to send message to.
+    ///
+    /// This is not used, because meta-service does not store node info in membership.
     target_node: MembershipNode,
 
-    /// A counter to send snapshot via v0 API.
-    ///
-    /// v0 API should only be used during upgrading a meta cluster.
-    /// During this period, i.e., this counter is `>0`,
-    /// try to send via v0 if the remote is not upgraded.
-    /// When this counter reaches 0, start sending via v1 API.
-    install_snapshot_via_v0: u64,
+    /// The endpoint of the target node.
+    endpoint: Endpoint,
 
     sto: RaftStore,
 
@@ -193,7 +191,7 @@ pub struct NetworkConnection {
 impl NetworkConnection {
     #[logcall::logcall(err = "debug")]
     #[minitrace::trace]
-    pub async fn make_client(&self) -> Result<(RaftClient, Endpoint), Unreachable> {
+    pub async fn make_client(&mut self) -> Result<RaftClient, Unreachable> {
         let target = self.target;
 
         let endpoint = self
@@ -215,7 +213,9 @@ impl NetworkConnection {
                 let client = RaftClientApi::new(target, endpoint.clone(), channel);
                 debug!("connected: target={}: {}", target, addr);
 
-                Ok((client, endpoint))
+                self.endpoint = endpoint;
+
+                Ok(client)
             }
             Err(err) => {
                 raft_metrics::network::incr_connect_failure(&target, &endpoint.to_string());
@@ -296,22 +296,41 @@ impl NetworkConnection {
         policy.chain(zero)
     }
 
-    /// Convert gRPC status to `RPCError`
-    fn status_to_unreachable<E>(
+    fn parse_grpc_resp<R, E>(
         &self,
-        status: tonic::Status,
-        endpoint: Endpoint,
-    ) -> RPCError<RaftError<E>>
+        grpc_res: Result<tonic::Response<RaftReply>, tonic::Status>,
+    ) -> Result<R, RPCError<RaftError<E>>>
     where
+        R: serde::de::DeserializeOwned + 'static,
+        E: serde::de::DeserializeOwned + 'static,
         E: std::error::Error,
     {
+        // Return status error
+        let resp = grpc_res.map_err(|e| self.status_to_unreachable(e))?;
+
+        // Parse serialized response into `Result<RaftReply.data, RaftReply.error>`
+        let raft_res = GrpcHelper::parse_raft_reply::<R, E>(resp).map_err(|serde_err| {
+            new_net_err(&serde_err, || {
+                let t = std::any::type_name::<R>();
+                format!("parse reply for {}", t)
+            })
+        })?;
+
+        // Wrap RaftError with RPCError
+        raft_res.map_err(|e| self.to_rpc_err(e))
+    }
+
+    /// Convert gRPC status to `RPCError`
+    fn status_to_unreachable<E>(&self, status: tonic::Status) -> RPCError<RaftError<E>>
+    where E: std::error::Error {
         warn!(
             "target={}, endpoint={} gRPC error: {:?}",
-            self.target, endpoint, status
+            self.target, self.endpoint, status
         );
 
         let any_err = AnyError::new(&status)
-            .add_context(|| format!("gRPC target={}, endpoint={}", self.target, endpoint));
+            .add_context(|| format!("gRPC target={}, endpoint={}", self.target, self.endpoint));
+
         RPCError::Unreachable(Unreachable::new(&any_err))
     }
 }
@@ -331,7 +350,7 @@ impl RaftNetwork<TypeConfig> for NetworkConnection {
             "send_append_entries",
         );
 
-        let (mut client, endpoint) = self.make_client().await?;
+        let mut client = self.make_client().await?;
 
         let raft_req = self.new_append_entries_raft_req(&rpc)?;
         let req = GrpcHelper::traced_req(raft_req);
@@ -348,12 +367,7 @@ impl RaftNetwork<TypeConfig> for NetworkConnection {
             self.target, grpc_res
         );
 
-        let resp = grpc_res.map_err(|e| self.status_to_unreachable(e, endpoint))?;
-
-        let raft_res = GrpcHelper::parse_raft_reply(resp)
-            .map_err(|serde_err| new_net_err(&serde_err, || "parse append_entries reply"))?;
-
-        raft_res.map_err(|e| self.to_rpc_err(e))
+        self.parse_grpc_resp(grpc_res)
     }
 
     #[logcall::logcall(err = "debug")]
@@ -390,68 +404,28 @@ impl RaftNetwork<TypeConfig> for NetworkConnection {
 
         let _g = snapshot_send_inflight(self.target).counter_guard();
 
-        let (mut client, endpoint) = self.make_client().await?;
+        let mut client = self.make_client().await?;
 
         let bytes = rpc.data.len() as u64;
         raft_metrics::network::incr_sendto_bytes(&self.target, bytes);
 
-        // Try send via `v1` API, if the remote peer does not provide `v1` API,
-        // revert to `v0` API.
-        let v1_res = if self.install_snapshot_via_v0 == 0 {
-            // Send via v1 API
-
-            let v1_req = SnapshotChunkRequest::new_v1(rpc.clone());
-            let req = databend_common_tracing::inject_span_to_tonic_request(v1_req);
-            let res = client
-                .install_snapshot_v1(req)
-                .timed(observe_snapshot_send_spent(self.target))
-                .await;
-
-            if is_unimplemented(&res) {
-                warn!(
-                    "target={} does not support install_snapshot_v1 API, fallback to v0 API for next 10 times",
-                    self.target
-                );
-                self.install_snapshot_via_v0 = 10;
-                None
-            } else {
-                Some(res)
-            }
-        } else {
-            None
-        };
+        let v1_req = SnapshotChunkRequest::new_v1(rpc);
+        let req = databend_common_tracing::inject_span_to_tonic_request(v1_req);
 
-        let grpc_res = if let Some(v1_res) = v1_res {
-            v1_res
-        } else {
-            // Via v1 API is not tried or failed,
-            // Send via v0 API
-
-            self.install_snapshot_via_v0 -= 1;
-
-            let req = databend_common_tracing::inject_span_to_tonic_request(rpc.clone());
-            client
-                .install_snapshot(req)
-                .timed(observe_snapshot_send_spent(self.target))
-                .await
-        };
+        let grpc_res = client
+            .install_snapshot_v1(req)
+            .timed(observe_snapshot_send_spent(self.target))
+            .await;
 
         info!(
             "install_snapshot resp target={}: {:?}",
             self.target, grpc_res
         );
 
-        let resp = grpc_res.map_err(|e| {
-            self.report_metrics_snapshot(false);
-            self.status_to_unreachable(e, endpoint)
-        })?;
-
-        let raft_res = GrpcHelper::parse_raft_reply(resp)
-            .map_err(|serde_err| new_net_err(&serde_err, || "parse install_snapshot reply"))?;
-
-        self.report_metrics_snapshot(raft_res.is_ok());
+        let res = self.parse_grpc_resp(grpc_res);
+        self.report_metrics_snapshot(res.is_ok());
 
-        raft_res.map_err(|e| self.to_rpc_err(e))
+        res
     }
 
     #[logcall::logcall(err = "debug")]
@@ -463,7 +437,7 @@ impl RaftNetwork<TypeConfig> for NetworkConnection {
     ) -> Result<VoteResponse, RPCError<RaftError>> {
         info!(id = self.id, target = self.target, rpc = rpc.summary(); "send_vote");
 
-        let (mut client, endpoint) = self.make_client().await?;
+        let mut client = self.make_client().await?;
 
         let raft_req = GrpcHelper::encode_raft_request(&rpc).map_err(|e| Unreachable::new(&e))?;
 
@@ -475,12 +449,7 @@ impl RaftNetwork<TypeConfig> for NetworkConnection {
         let grpc_res = client.vote(req).await;
         info!("vote: resp from target={} {:?}", self.target, grpc_res);
 
-        let resp = grpc_res.map_err(|e| self.status_to_unreachable(e, endpoint))?;
-
-        let raft_res = GrpcHelper::parse_raft_reply(resp)
-            .map_err(|serde_err| new_net_err(&serde_err, || "parse vote reply"))?;
-
-        raft_res.map_err(|e| self.to_rpc_err(e))
+        self.parse_grpc_resp(grpc_res)
     }
 
     /// When a `Unreachable` error is returned from the `Network`,
@@ -508,10 +477,10 @@ impl RaftNetworkFactory<TypeConfig> for Network {
             id: self.sto.id,
             target,
             target_node: node.clone(),
-            install_snapshot_via_v0: 0,
             sto: self.sto.clone(),
             conn_pool: self.conn_pool.clone(),
             backoff: self.backoff.clone(),
+            endpoint: Default::default(),
         }
     }
 }
@@ -541,14 +510,3 @@ fn observe_snapshot_send_spent(target: NodeId) -> impl Fn(Duration, Duration) {
 fn snapshot_send_inflight(target: NodeId) -> impl FnMut(i64) {
     move |i: i64| raft_metrics::network::incr_snapshot_sendto_inflight(&target, i)
 }
-
-/// Return true if it IS an error and the error code is Unimplemented.
-///
-/// Return false if it is NOT an error or the error code is NOT Unimplemented.
-fn is_unimplemented<T>(res: &Result<T, tonic::Status>) -> bool {
-    if let Err(status) = res {
-        status.code() == tonic::Code::Unimplemented
-    } else {
-        false
-    }
-}

src/meta/service/src/version.rs

@@ -99,6 +99,7 @@ pub static MIN_META_SEMVER: Version = Version::new(0, 9, 41);
 pub(crate) mod raft {
     pub(crate) mod server {
         use feature_set::add_provide;
+        use feature_set::del_provide;
         use feature_set::Action;
         use feature_set::Provide;
 
@@ -113,6 +114,8 @@ pub(crate) mod raft {
             add_provide(("install_snapshot", 0), "2023-02-16", (0,  9,  41)),
             add_provide(("install_snapshot", 1), "2023-11-16", (1,  2, 212)),
             add_provide(("install_snapshot", 2), "2024-05-06", (1,  2, 453)),
+            // TODO: fix date and version when merged.
+            del_provide(("install_snapshot", 0), "2024-05-20", (1,  2, 476)),
         ];
 
         /// The client features that raft server depends on.
@@ -125,6 +128,7 @@ pub(crate) mod raft {
     pub(crate) mod client {
         use feature_set::add_optional;
         use feature_set::add_require;
+        use feature_set::del_require;
         use feature_set::Action;
         use feature_set::Require;
 
@@ -138,6 +142,9 @@ pub(crate) mod raft {
             add_require( ("append",           0), "2023-02-16", (0,  9,  41)),
             add_require( ("install_snapshot", 0), "2023-02-16", (0,  9,  41)),
             add_optional(("install_snapshot", 1), "2023-11-16", (1,  2, 212)),
+            // TODO: fix date and version when merged.
+            add_require( ("install_snapshot", 1), "2023-05-20", (1,  2, 476)),
+            del_require( ("install_snapshot", 0), "2024-05-20", (1,  2, 476)),
         ];
 
         /// Feature set provided by raft client.

src/meta/types/proto/meta.proto

@@ -243,7 +243,6 @@ service RaftService {
   // raft RPC
 
   rpc AppendEntries(RaftRequest) returns (RaftReply);
-  rpc InstallSnapshot(RaftRequest) returns (RaftReply);
   rpc InstallSnapshotV1(SnapshotChunkRequest) returns (RaftReply);
   rpc InstallSnapshotV2(stream SnapshotChunkRequestV2) returns (RaftReply);
   rpc Vote(RaftRequest) returns (RaftReply);

src/meta/types/src/grpc_helper.rs

@@ -124,7 +124,7 @@ impl GrpcHelper {
     ) -> Result<Result<T, RaftError<E>>, serde_json::Error>
     where
         T: serde::de::DeserializeOwned,
-        E: serde::Serialize + serde::de::DeserializeOwned,
+        E: serde::de::DeserializeOwned,
     {
         let raft_reply = reply.into_inner();