xmtp_api_grpc/
error.rs

1use thiserror::Error;
2use xmtp_common::ErrorCode;
3use xmtp_proto::ConversionError;
4
5#[derive(Debug, Error, ErrorCode)]
6pub enum GrpcBuilderError {
7    /// Missing app version.
8    ///
9    /// App version not set on builder. Not retryable.
10    #[error("app version required to create client")]
11    MissingAppVersion,
12    /// Missing LibXMTP version.
13    ///
14    /// Core library version not set. Not retryable.
15    #[error("libxmtp core library version required to create client")]
16    MissingLibxmtpVersion,
17    /// Missing host URL.
18    ///
19    /// Host URL not set on builder. Not retryable.
20    #[error("host url required to create client")]
21    MissingHostUrl,
22    /// Metadata error.
23    ///
24    /// Invalid gRPC metadata value. Not retryable.
25    #[error(transparent)]
26    Metadata(#[from] tonic::metadata::errors::InvalidMetadataValue),
27    /// Invalid URI.
28    ///
29    /// URI is malformed. Not retryable.
30    #[error("Invalid URI during channel creation")]
31    InvalidUri(#[from] http::uri::InvalidUri),
32    /// URL parse error.
33    ///
34    /// URL string is malformed. Not retryable.
35    #[error(transparent)]
36    Url(#[from] url::ParseError),
37    /// Transport error.
38    ///
39    /// gRPC transport creation failed (native only). Not retryable.
40    #[cfg(not(all(target_family = "wasm", target_os = "unknown")))]
41    #[error(transparent)]
42    Transport(#[from] tonic::transport::Error),
43}
44
45#[derive(Debug, Error, ErrorCode)]
46pub enum GrpcError {
47    /// Invalid URI.
48    ///
49    /// URI for channel creation is malformed. Not retryable.
50    #[error("Invalid URI during channel creation")]
51    InvalidUri(#[from] http::uri::InvalidUri),
52    /// Metadata error.
53    ///
54    /// Invalid gRPC metadata value. Not retryable.
55    #[error(transparent)]
56    Metadata(#[from] tonic::metadata::errors::InvalidMetadataValue),
57    /// gRPC status error.
58    ///
59    /// Retryability depends on the gRPC status code.
60    #[error("{0}")]
61    Status(#[from] tonic::Status),
62    /// Not found.
63    ///
64    /// Requested resource not found, empty, or proto conversion failed. Not retryable.
65    #[error("{0} not found/empty")]
66    NotFound(String),
67    /// Unexpected payload.
68    ///
69    /// Payload not expected in response. Not retryable.
70    #[error("Payload not expected")]
71    UnexpectedPayload,
72    /// Missing payload.
73    ///
74    /// Expected payload not in response. Not retryable.
75    #[error("payload is missing")]
76    MissingPayload,
77    #[error(transparent)]
78    #[error_code(inherit)]
79    Proto(#[from] xmtp_proto::ProtoError),
80    /// Decode error.
81    ///
82    /// Protobuf decoding failed. Not retryable.
83    #[error(transparent)]
84    Decode(#[from] prost::DecodeError),
85    /// Unreachable.
86    ///
87    /// Infallible error. Not retryable.
88    #[error("unreachable (Infallible)")]
89    Unreachable,
90    /// Transport error.
91    ///
92    /// gRPC transport layer error (native only). Retryable.
93    #[cfg(not(all(target_family = "wasm", target_os = "unknown")))]
94    #[error(transparent)]
95    Transport(#[from] tonic::transport::Error),
96}
97
98impl From<ConversionError> for GrpcError {
99    fn from(error: ConversionError) -> Self {
100        GrpcError::NotFound(error.to_string())
101    }
102}
103
104impl GrpcError {
105    /// Return whether the server reports that the RPC is not implemented.
106    pub fn is_unimplemented(&self) -> bool {
107        matches!(self, Self::Status(status) if status.code() == tonic::Code::Unimplemented)
108    }
109}
110
111impl xmtp_common::retry::RetryableError for GrpcError {
112    fn is_retryable(&self) -> bool {
113        use tonic::Code;
114
115        match self {
116            Self::Status(status) => match status.code() {
117                Code::InvalidArgument
118                | Code::OutOfRange
119                | Code::Unimplemented
120                | Code::Aborted
121                | Code::NotFound
122                | Code::AlreadyExists
123                | Code::FailedPrecondition
124                | Code::PermissionDenied
125                | Code::Unauthenticated
126                | Code::DataLoss
127                | Code::Ok => false,
128                Code::Cancelled => is_transport_cancellation(status),
129                Code::Unavailable
130                | Code::ResourceExhausted
131                | Code::DeadlineExceeded
132                | Code::Unknown
133                | Code::Internal => true,
134            },
135            #[cfg(not(all(target_family = "wasm", target_os = "unknown")))]
136            Self::Transport(_) => true,
137            Self::Proto(_)
138            | Self::InvalidUri(_)
139            | Self::Metadata(_)
140            | Self::NotFound(_)
141            | Self::UnexpectedPayload
142            | Self::MissingPayload
143            | Self::Decode(_)
144            | Self::Unreachable => false,
145        }
146    }
147}
148
149/// Hyper connection closure and Tonic's local timeout can produce Cancelled.
150/// Keep explicit RPC cancellation permanent when these typed causes are absent.
151fn is_transport_cancellation(status: &tonic::Status) -> bool {
152    let mut source = std::error::Error::source(status);
153    while let Some(error) = source {
154        if error.is::<tonic::TimeoutExpired>()
155            || error
156                .downcast_ref::<hyper::Error>()
157                .is_some_and(hyper::Error::is_canceled)
158        {
159            return true;
160        }
161        source = error.source();
162    }
163    false
164}
165
166#[cfg(test)]
167mod tests {
168    use super::GrpcError;
169    use tonic::{Code, Status};
170    use xmtp_common::RetryableError;
171
172    #[rstest::rstest]
173    #[case(Code::Ok, false)]
174    #[case(Code::Cancelled, false)]
175    #[case(Code::Unknown, true)]
176    #[case(Code::InvalidArgument, false)]
177    #[case(Code::DeadlineExceeded, true)]
178    #[case(Code::NotFound, false)]
179    #[case(Code::AlreadyExists, false)]
180    #[case(Code::PermissionDenied, false)]
181    #[case(Code::ResourceExhausted, true)]
182    #[case(Code::FailedPrecondition, false)]
183    #[case(Code::Aborted, false)]
184    #[case(Code::OutOfRange, false)]
185    #[case(Code::Unimplemented, false)]
186    #[case(Code::Internal, true)]
187    #[case(Code::Unavailable, true)]
188    #[case(Code::DataLoss, false)]
189    #[case(Code::Unauthenticated, false)]
190    #[xmtp_common::test(unwrap_try = true)]
191    async fn retry_by_status_code(#[case] code: Code, #[case] retryable: bool) {
192        for message in ["", "UNAVAILABLE", "INVALID_ARGUMENT", "request too large"] {
193            let error = GrpcError::Status(Status::new(code, message));
194            assert_eq!(error.is_retryable(), retryable, "{code:?}: {message}");
195        }
196    }
197
198    #[xmtp_common::test(unwrap_try = true)]
199    fn explicit_rpc_cancellation_is_not_retryable() {
200        use xmtp_proto::api::{ApiClientError, NetworkError, grpc_status};
201
202        for message in [
203            "",
204            "operation was canceled",
205            "connection closed",
206            "Timeout expired",
207        ] {
208            let error = NetworkError::new(ApiClientError::client(GrpcError::Status(
209                Status::cancelled(message),
210            )));
211            assert_eq!(grpc_status(&error)?.code(), Code::Cancelled);
212            assert!(!error.is_retryable());
213        }
214        let mut status = Status::cancelled("operation was canceled");
215        status.set_source(std::sync::Arc::new(std::io::Error::other(
216            "connection closed",
217        )));
218        assert!(!GrpcError::Status(status).is_retryable());
219    }
220
221    #[cfg(not(all(target_family = "wasm", target_os = "unknown")))]
222    #[xmtp_common::test(unwrap_try = true)]
223    async fn local_timeout_is_retryable_through_tonic_and_client_wrappers() {
224        use hyper_util::rt::TokioIo;
225        use xmtp_common::time::{Duration, timeout};
226        use xmtp_proto::api::{ApiClientError, NetworkError, grpc_status};
227
228        for wrapped in [false, true] {
229            let status = if wrapped {
230                let connector = tower::service_fn(move |_| {
231                    std::future::ready(Err::<TokioIo<tokio::io::DuplexStream>, _>(
232                        tonic::TimeoutExpired(()),
233                    ))
234                });
235                let endpoint = tonic::transport::Endpoint::from_static("http://unused.invalid");
236                let transport = timeout(
237                    Duration::from_secs(5),
238                    endpoint.connect_with_connector(connector),
239                )
240                .await?
241                .expect_err("the connector returns the local timeout");
242                // Keep the actual transport wrapper and its typed timeout cause.
243                let mut status = Status::cancelled("retained transport cause");
244                status.set_source(std::sync::Arc::new(transport));
245                status
246            } else {
247                Status::from_error(Box::new(tonic::TimeoutExpired(())))
248            };
249            assert_eq!(status.code(), Code::Cancelled);
250            let error = GrpcError::Status(status);
251            assert!(error.is_retryable());
252            let error = ApiClientError::client(error);
253            assert!(error.is_retryable());
254            let error = NetworkError::new(error);
255            assert_eq!(grpc_status(&error)?.code(), Code::Cancelled);
256            assert!(error.is_retryable());
257        }
258    }
259
260    #[cfg(not(all(target_family = "wasm", target_os = "unknown")))]
261    #[xmtp_common::test(unwrap_try = true)]
262    async fn cancelled_hyper_request_is_retryable_through_tonic_and_client_wrappers() {
263        use http_body_util::Empty;
264        use hyper_util::rt::TokioIo;
265        use prost::bytes::Bytes;
266        use xmtp_common::time::{Duration, timeout};
267        use xmtp_proto::api::{ApiClientError, NetworkError, grpc_status};
268
269        type Body = Empty<Bytes>;
270        for wrapped in [false, true] {
271            let (io, _peer) = tokio::io::duplex(64);
272            let (mut sender, connection) =
273                hyper::client::conn::http1::handshake::<_, Body>(TokioIo::new(io)).await?;
274            drop(connection);
275            let cancelled = sender
276                .send_request(http::Request::new(Body::new()))
277                .await
278                .expect_err("the dropped connection cannot send the request");
279            assert!(cancelled.is_canceled());
280
281            let status = if wrapped {
282                let mut cancelled = Some(cancelled);
283                let connector = tower::service_fn(move |_| {
284                    std::future::ready(Err::<TokioIo<tokio::io::DuplexStream>, _>(
285                        cancelled.take().expect("one connection attempt"),
286                    ))
287                });
288                let endpoint = tonic::transport::Endpoint::from_static("http://unused.invalid");
289                let transport = timeout(
290                    Duration::from_secs(5),
291                    endpoint.connect_with_connector(connector),
292                )
293                .await?
294                .expect_err("the connector returns the cancelled request");
295                // Connection-open failures map to Unavailable. Attach only their typed cause.
296                let mut status = Status::cancelled("retained transport cause");
297                status.set_source(std::sync::Arc::new(transport));
298                status
299            } else {
300                Status::from_error(Box::new(cancelled))
301            };
302            assert_eq!(status.code(), Code::Cancelled);
303            let error = GrpcError::Status(status);
304            assert!(error.is_retryable());
305            let error = ApiClientError::client(error);
306            assert!(error.is_retryable());
307            let error = NetworkError::new(error);
308            assert_eq!(grpc_status(&error)?.code(), Code::Cancelled);
309            assert!(error.is_retryable());
310        }
311    }
312}
313
314#[cfg(test)]
315mod status_sources {
316    use super::GrpcError;
317    use xmtp_proto::api::{ApiClientError, grpc_status};
318
319    #[xmtp_common::test(unwrap_try = true)]
320    fn typed_status_survives_client_error_wrappers() {
321        let error =
322            ApiClientError::client(GrpcError::Status(tonic::Status::aborted("unchanged text")));
323        assert_eq!(grpc_status(&error)?.code(), tonic::Code::Aborted);
324        let error = ApiClientError::other(GrpcError::Status(tonic::Status::out_of_range(
325            "unchanged text",
326        )));
327        assert_eq!(grpc_status(&error)?.code(), tonic::Code::OutOfRange);
328    }
329}
xmtp_api_grpc/error.rs

xmtp_api_grpc/
error.rs
