risingwave_rpc_client/
error.rs

1// Copyright 2025 RisingWave Labs
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7//     http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15use risingwave_common::util::meta_addr::MetaAddressStrategyParseError;
16use risingwave_error::tonic::TonicStatusWrapperExt as _;
17use thiserror::Error;
18use thiserror_ext::Construct;
19
20pub type Result<T, E = RpcError> = std::result::Result<T, E>;
21
22// Re-export these types as they're commonly used together with `RpcError`.
23pub use risingwave_error::tonic::{ToTonicStatus, TonicStatusWrapper};
24
25#[derive(Error, Debug, Construct)]
26pub enum RpcError {
27    #[error(transparent)]
28    TransportError(Box<tonic::transport::Error>),
29
30    #[error(transparent)]
31    GrpcStatus(
32        #[from]
33        // Typically it does not have a backtrace,
34        // but this is to let `thiserror` generate `provide` implementation to make `Extra` work.
35        // See `risingwave_error::tonic::extra`.
36        #[backtrace]
37        Box<TonicStatusWrapper>,
38    ),
39
40    #[error(transparent)]
41    MetaAddressParse(#[from] MetaAddressStrategyParseError),
42
43    #[error(transparent)]
44    Internal(
45        #[from]
46        #[backtrace]
47        anyhow::Error,
48    ),
49}
50
51// TODO: use `thiserror_ext::Box`
52static_assertions::const_assert_eq!(std::mem::size_of::<RpcError>(), 32);
53
54impl From<tonic::transport::Error> for RpcError {
55    fn from(e: tonic::transport::Error) -> Self {
56        RpcError::TransportError(Box::new(e))
57    }
58}
59
60/// Intentionally not implemented to enforce using `RpcError::from_xxx_status`, so that
61/// the service name can always be included in the error message.
62impl !From<tonic::Status> for RpcError {}
63
64macro_rules! impl_from_status {
65    ($($service:ident),* $(,)?) => {
66        paste::paste! {
67            impl RpcError {
68                $(
69                    #[doc = "Convert a gRPC status from " $service " service into an [`RpcError`]."]
70                    pub fn [<from_ $service _status>](s: tonic::Status) -> Self {
71                        Box::new(s.with_client_side_service_name(stringify!($service))).into()
72                    }
73                )*
74            }
75        }
76    };
77}
78
79impl_from_status!(stream, batch, meta, compute, compactor, connector, frontend);
80
81impl RpcError {
82    /// Returns `true` if the error is a connection error. Typically used to determine if
83    /// the error is transient and can be retried.
84    pub fn is_connection_error(&self) -> bool {
85        match self {
86            RpcError::TransportError(_) => true,
87            RpcError::GrpcStatus(status) => matches!(
88                status.inner().code(),
89                tonic::Code::Unavailable // server not started
90                 | tonic::Code::Unknown // could be transport error
91                 | tonic::Code::Unimplemented // meta leader service not started
92            ),
93            RpcError::MetaAddressParse(_) => false,
94            RpcError::Internal(anyhow) => anyhow
95                .downcast_ref::<Self>() // this skips all contexts attached to the error
96                .is_some_and(Self::is_connection_error),
97        }
98    }
99}
risingwave_rpc_client/error.rs

risingwave_rpc_client/
error.rs
