Classify ProposalDropped as Unavailable Tonic status

The raft metadata store does not accept new proposals if there is
no known leader. In this situation, request failed with an internal
ProposalDropped error. This commit changes the behavior so that a
ProposalDropped error will be translated into an unavailable Tonic
status. That way, the request will get automatically retried.

crates/metadata-store/src/grpc/handler.rs

@@ -132,7 +132,8 @@ impl MetadataStoreSvc for MetadataStoreHandler {
 impl From<RequestError> for Status {
     fn from(err: RequestError) -> Self {
         match err {
-            RequestError::FailedPrecondition(msg) => Status::failed_precondition(msg.to_string()),
+            RequestError::FailedPrecondition(err) => Status::failed_precondition(err.to_string()),
+            RequestError::Unavailable(err) => Status::unavailable(err.to_string()),
             err => Status::internal(err.to_string()),
         }
     }

crates/metadata-store/src/lib.rs

@@ -32,7 +32,9 @@ pub type RequestReceiver = mpsc::Receiver<MetadataStoreRequest>;
 #[derive(Debug, thiserror::Error)]
 pub enum RequestError {
     #[error("internal error: {0}")]
-    Internal(#[from] GenericError),
+    Internal(GenericError),
+    #[error("service currently unavailable: {0}")]
+    Unavailable(GenericError),
     #[error("failed precondition: {0}")]
     FailedPrecondition(#[from] PreconditionViolation),
     #[error("invalid argument: {0}")]

crates/metadata-store/src/raft/store.rs

@@ -112,7 +112,7 @@ impl RaftMetadataStore {
                         .map_err(Into::into)
                         .and_then(|request| self.raw_node
                             .propose(vec![], request)
-                            .map_err(|err| RequestError::Internal(err.into()))) {
+                            .map_err(|err| RequestError::from(err))) {
                         info!("Failed processing request: {err}");
                         callback.fail(err);
                         continue;
@@ -527,3 +527,12 @@ enum RequestKind {
         precondition: Precondition,
     },
 }
+
+impl From<raft::Error> for RequestError {
+    fn from(value: raft::Error) -> Self {
+        match value {
+            err @ raft::Error::ProposalDropped => RequestError::Unavailable(err.into()),
+            err => RequestError::Internal(err.into()),
+        }
+    }
+}