fix: Respond with SnapshotMismatch error instead of internal error (#15391)

This fix updates OpenRaft to version 0.9.8 to address an issue where,
upon receiving a `SnapshotMismatch` error, the snapshot should be resent
from the beginning. This ensures that the leader can correctly reset and
reinitiate the snapshot transmission process.

And meta-service should return `SnapshotMismatch` error instead of
returning an internal gRPC error.

Author: drmingdrmer

Cargo.lock

@@ -123,7 +123,7 @@ sled = { git = "https://github.com/datafuse-extras/sled", tag = "v0.34.7-datafus
 
 # openraft for debugging
 #openraft = { git = "https://github.com/drmingdrmer/openraft", branch = "release-0.9", features = [
-openraft = { version = "0.9.7", features = [
+openraft = { version = "0.9.8", features = [
     "serde",
     "tracing-log",
     "generic-snapshot-data",

Cargo.toml

@@ -82,18 +82,14 @@ impl RaftServiceImpl {
 
         let is_req = GrpcHelper::parse_req(request)?;
 
-        let resp = self
+        let res = self
             .receive_chunked_snapshot(is_req)
             .timed(observe_snapshot_recv_spent(&addr))
-            .await
-            .map_err(GrpcHelper::internal_err);
+            .await;
 
-        raft_metrics::network::incr_snapshot_recvfrom_result(addr.clone(), resp.is_ok());
+        raft_metrics::network::incr_snapshot_recvfrom_result(addr.clone(), res.is_ok());
 
-        match resp {
-            Ok(resp) => GrpcHelper::ok_response(resp),
-            Err(e) => Err(e),
-        }
+        GrpcHelper::make_grpc_result(res)
     }
 
     async fn do_install_snapshot_v1(
@@ -122,18 +118,14 @@ impl RaftServiceImpl {
             done: chunk.done,
         };
 
-        let resp = self
+        let res = self
             .receive_chunked_snapshot(install_snapshot_req)
             .timed(observe_snapshot_recv_spent(&addr))
-            .await
-            .map_err(GrpcHelper::internal_err);
+            .await;
 
-        raft_metrics::network::incr_snapshot_recvfrom_result(addr.clone(), resp.is_ok());
+        raft_metrics::network::incr_snapshot_recvfrom_result(addr.clone(), res.is_ok());
 
-        match resp {
-            Ok(resp) => GrpcHelper::ok_response(resp),
-            Err(e) => Err(e),
-        }
+        GrpcHelper::make_grpc_result(res)
     }
 
     /// Receive a chunk based snapshot from the leader.
@@ -227,7 +219,7 @@ impl RaftService for RaftServiceImpl {
                 .await
                 .map_err(GrpcHelper::internal_err)?;
 
-            GrpcHelper::ok_response(resp)
+            GrpcHelper::ok_response(&resp)
         }
         .in_span(root)
         .await
@@ -260,7 +252,7 @@ impl RaftService for RaftServiceImpl {
 
             let resp = raft.vote(v_req).await.map_err(GrpcHelper::internal_err)?;
 
-            GrpcHelper::ok_response(resp)
+            GrpcHelper::ok_response(&resp)
         }
         .in_span(root)
         .await

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

@@ -12,17 +12,25 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+use databend_common_meta_raft_store::state_machine::MetaSnapshotId;
 use databend_common_meta_sled_store::openraft::LogIdOptionExt;
 use databend_common_meta_sled_store::openraft::ServerState;
+use databend_common_meta_types::protobuf::SnapshotChunkRequest;
 use databend_common_meta_types::Cmd;
+use databend_common_meta_types::InstallSnapshotRequest;
 use databend_common_meta_types::LogEntry;
 use databend_common_meta_types::SeqV;
+use databend_common_meta_types::SnapshotMeta;
+use databend_common_meta_types::StoredMembership;
 use databend_common_meta_types::UpsertKV;
+use databend_common_meta_types::Vote;
 use databend_meta::meta_service::MetaNode;
 use log::info;
+use maplit::btreeset;
 use test_harness::test;
 
 use crate::testing::meta_service_test_harness;
+use crate::tests::meta_node::start_meta_node_cluster;
 use crate::tests::meta_node::start_meta_node_non_voter;
 use crate::tests::meta_node::timeout;
 use crate::tests::service::MetaSrvTestContext;
@@ -135,3 +143,36 @@ async fn test_meta_node_snapshot_replication() -> anyhow::Result<()> {
 
     Ok(())
 }
+
+#[test(harness = meta_service_test_harness)]
+#[minitrace::trace]
+async fn test_raft_service_snapshot_id_mismatch() -> anyhow::Result<()> {
+    // Test SnapshotIdMismatch error should be responded.
+
+    let (mut _nlog, mut tcs) = start_meta_node_cluster(btreeset![0], btreeset![]).await?;
+    let tc0 = tcs.remove(0);
+
+    let mut client0 = tc0.raft_client().await?;
+    let mut r1 = InstallSnapshotRequest {
+        vote: Vote::new_committed(10, 2),
+        meta: SnapshotMeta {
+            last_log_id: None,
+            last_membership: StoredMembership::default(),
+            snapshot_id: MetaSnapshotId::new(None, 1).to_string(),
+        },
+        offset: 0,
+        data: vec![1, 2, 3],
+        done: false,
+    };
+
+    let req = SnapshotChunkRequest::new_v1(r1.clone());
+    client0.install_snapshot_v1(req).await?;
+
+    r1.meta.snapshot_id = MetaSnapshotId::new(None, 2).to_string();
+    r1.offset = 3;
+    let req = SnapshotChunkRequest::new_v1(r1);
+    let res = client0.install_snapshot_v1(req).await;
+    println!("res: {:?}", res);
+
+    Ok(())
+}

src/meta/service/tests/it/meta_node/meta_node_replication.rs

@@ -219,7 +219,7 @@ impl MetaSrvTestContext {
         let addr = self.config.raft_config.raft_api_addr().await?;
 
         // retry 3 times until server starts listening.
-        for _ in 0..4 {
+        for _ in 0..3 {
             let client = RaftServiceClient::connect(format!("http://{}", addr)).await;
             match client {
                 Ok(x) => return Ok(x),

src/meta/service/tests/it/tests/service.rs

@@ -144,17 +144,48 @@ impl GrpcHelper {
         Self::parse(&raft_req.data)
     }
 
+    /// Make a gRPC response result with Ok, ApiError or internal error.
+    pub fn make_grpc_result<D, E>(
+        res: Result<D, RaftError<E>>,
+    ) -> Result<tonic::Response<RaftReply>, tonic::Status>
+    where
+        D: serde::Serialize + 'static,
+        E: serde::Serialize + Error + 'static,
+    {
+        match res {
+            Ok(resp) => GrpcHelper::ok_response(&resp),
+            Err(e) => {
+                if e.api_error().is_some() {
+                    GrpcHelper::err_response(&e)
+                } else {
+                    Err(GrpcHelper::internal_err(e))
+                }
+            }
+        }
+    }
+
     /// Create an Ok response for raft API.
-    pub fn ok_response<D>(d: D) -> Result<tonic::Response<RaftReply>, tonic::Status>
-    where D: serde::Serialize {
-        let data = serde_json::to_string(&d).expect("fail to serialize resp");
+    pub fn ok_response<D>(d: &D) -> Result<tonic::Response<RaftReply>, tonic::Status>
+    where D: serde::Serialize + 'static {
+        let data = serde_json::to_string(d).expect("fail to serialize resp");
         let reply = RaftReply {
             data,
             error: "".to_string(),
         };
         Ok(tonic::Response::new(reply))
     }
 
+    /// Create an Ok response contains API error.
+    pub fn err_response<E>(e: &E) -> Result<tonic::Response<RaftReply>, tonic::Status>
+    where E: serde::Serialize + 'static {
+        let error = serde_json::to_string(e).expect("fail to serialize response error");
+        let reply = RaftReply {
+            data: "".to_string(),
+            error,
+        };
+        Ok(tonic::Response::new(reply))
+    }
+
     /// Parse string and decode it into required type.
     pub fn parse<T>(s: &str) -> Result<T, tonic::Status>
     where T: serde::de::DeserializeOwned {