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/// A closed Hyper connection can cancel a request that was not sent.
150/// Keep explicit RPC cancellation permanent when this transport cause is 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
155            .downcast_ref::<hyper::Error>()
156            .is_some_and(hyper::Error::is_canceled)
157        {
158            return true;
159        }
160        source = error.source();
161    }
162    false
163}
164
165#[cfg(test)]
166mod tests {
167    use super::GrpcError;
168    use tonic::{Code, Status};
169    use xmtp_common::RetryableError;
170
171    #[rstest::rstest]
172    #[case(Code::Ok, false)]
173    #[case(Code::Cancelled, false)]
174    #[case(Code::Unknown, true)]
175    #[case(Code::InvalidArgument, false)]
176    #[case(Code::DeadlineExceeded, true)]
177    #[case(Code::NotFound, false)]
178    #[case(Code::AlreadyExists, false)]
179    #[case(Code::PermissionDenied, false)]
180    #[case(Code::ResourceExhausted, true)]
181    #[case(Code::FailedPrecondition, false)]
182    #[case(Code::Aborted, false)]
183    #[case(Code::OutOfRange, false)]
184    #[case(Code::Unimplemented, false)]
185    #[case(Code::Internal, true)]
186    #[case(Code::Unavailable, true)]
187    #[case(Code::DataLoss, false)]
188    #[case(Code::Unauthenticated, false)]
189    #[xmtp_common::test(unwrap_try = true)]
190    async fn retry_by_status_code(#[case] code: Code, #[case] retryable: bool) {
191        for message in ["", "UNAVAILABLE", "INVALID_ARGUMENT", "request too large"] {
192            let error = GrpcError::Status(Status::new(code, message));
193            assert_eq!(error.is_retryable(), retryable, "{code:?}: {message}");
194        }
195    }
196
197    #[xmtp_common::test(unwrap_try = true)]
198    fn explicit_rpc_cancellation_is_not_retryable() {
199        use xmtp_proto::api::{ApiClientError, NetworkError, grpc_status};
200
201        for message in ["", "operation was canceled", "connection closed"] {
202            let error = NetworkError::new(ApiClientError::client(GrpcError::Status(
203                Status::cancelled(message),
204            )));
205            assert_eq!(grpc_status(&error)?.code(), Code::Cancelled);
206            assert!(!error.is_retryable());
207        }
208        let mut status = Status::cancelled("operation was canceled");
209        status.set_source(std::sync::Arc::new(std::io::Error::other(
210            "connection closed",
211        )));
212        assert!(!GrpcError::Status(status).is_retryable());
213    }
214
215    #[cfg(not(all(target_family = "wasm", target_os = "unknown")))]
216    #[xmtp_common::test(unwrap_try = true)]
217    async fn cancelled_hyper_request_is_retryable_through_tonic_and_client_wrappers() {
218        use http_body_util::Empty;
219        use hyper_util::rt::TokioIo;
220        use prost::bytes::Bytes;
221        use xmtp_common::time::{Duration, timeout};
222        use xmtp_proto::api::{ApiClientError, NetworkError, grpc_status};
223
224        type Body = Empty<Bytes>;
225        for wrapped in [false, true] {
226            let (io, _peer) = tokio::io::duplex(64);
227            let (mut sender, connection) =
228                hyper::client::conn::http1::handshake::<_, Body>(TokioIo::new(io)).await?;
229            drop(connection);
230            let cancelled = sender
231                .send_request(http::Request::new(Body::new()))
232                .await
233                .expect_err("the dropped connection cannot send the request");
234            assert!(cancelled.is_canceled());
235
236            let status = if wrapped {
237                let mut cancelled = Some(cancelled);
238                let connector = tower::service_fn(move |_| {
239                    std::future::ready(Err::<TokioIo<tokio::io::DuplexStream>, _>(
240                        cancelled.take().expect("one connection attempt"),
241                    ))
242                });
243                let endpoint = tonic::transport::Endpoint::from_static("http://unused.invalid");
244                let transport = timeout(
245                    Duration::from_secs(5),
246                    endpoint.connect_with_connector(connector),
247                )
248                .await?
249                .expect_err("the connector returns the cancelled request");
250                // Connection-open failures map to Unavailable. Attach only their typed cause.
251                let mut status = Status::cancelled("retained transport cause");
252                status.set_source(std::sync::Arc::new(transport));
253                status
254            } else {
255                Status::from_error(Box::new(cancelled))
256            };
257            assert_eq!(status.code(), Code::Cancelled);
258            let error = GrpcError::Status(status);
259            assert!(error.is_retryable());
260            let error = ApiClientError::client(error);
261            assert!(error.is_retryable());
262            let error = NetworkError::new(error);
263            assert_eq!(grpc_status(&error)?.code(), Code::Cancelled);
264            assert!(error.is_retryable());
265        }
266    }
267}
268
269#[cfg(test)]
270mod status_sources {
271    use super::GrpcError;
272    use xmtp_proto::api::{ApiClientError, grpc_status};
273
274    #[xmtp_common::test(unwrap_try = true)]
275    fn typed_status_survives_client_error_wrappers() {
276        let error =
277            ApiClientError::client(GrpcError::Status(tonic::Status::aborted("unchanged text")));
278        assert_eq!(grpc_status(&error)?.code(), tonic::Code::Aborted);
279        let error = ApiClientError::other(GrpcError::Status(tonic::Status::out_of_range(
280            "unchanged text",
281        )));
282        assert_eq!(grpc_status(&error)?.code(), tonic::Code::OutOfRange);
283    }
284}
xmtp_api_grpc/error.rs

xmtp_api_grpc/
error.rs
