typedb_driver/common/
error.rs

1/*
2 * Licensed to the Apache Software Foundation (ASF) under one
3 * or more contributor license agreements.  See the NOTICE file
4 * distributed with this work for additional information
5 * regarding copyright ownership.  The ASF licenses this file
6 * to you under the Apache License, Version 2.0 (the
7 * "License"); you may not use this file except in compliance
8 * with the License.  You may obtain a copy of the License at
9 *
10 *   http://www.apache.org/licenses/LICENSE-2.0
11 *
12 * Unless required by applicable law or agreed to in writing,
13 * software distributed under the License is distributed on an
14 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
15 * KIND, either express or implied.  See the License for the
16 * specific language governing permissions and limitations
17 * under the License.
18 */
19
20use std::{collections::HashSet, error::Error as StdError, fmt};
21
22use tonic::{Code, Status};
23use tonic_types::StatusExt;
24
25use super::{RequestID, address::Address};
26
27macro_rules! error_messages {
28    {
29        $name:ident code: $code_pfx:literal, type: $message_pfx:literal,
30        $($error_name:ident $({ $($field:ident : $inner:ty),+ $(,)? })? = $code:literal: $body:literal),+ $(,)?
31    } => {
32        #[derive(Clone, Eq, PartialEq)]
33        pub enum $name {$(
34            $error_name$( { $($field: $inner),+ })?,
35        )*}
36
37        impl $name {
38            pub const PREFIX: &'static str = $code_pfx;
39
40            pub const fn code(&self) -> usize {
41                match self {$(
42                    Self::$error_name $({ $($field: _),+ })? => $code,
43                )*}
44            }
45
46            pub fn format_code(&self) -> String {
47                format!(concat!("[", $code_pfx, "{}{}]"), self.padding(), self.code())
48            }
49
50            pub fn message(&self) -> String {
51                match self {$(
52                    Self::$error_name $({$($field),+})? => format!($body $($(, $field = $field)+)?),
53                )*}
54            }
55
56            const fn max_code() -> usize {
57                let mut max = usize::MIN;
58                $(max = if $code > max { $code } else { max };)*
59                max
60            }
61
62            const fn num_digits(x: usize) -> usize {
63                if (x < 10) { 1 } else { 1 + Self::num_digits(x/10) }
64            }
65
66            const fn padding(&self) -> &'static str {
67                match Self::num_digits(Self::max_code()) - Self::num_digits(self.code()) {
68                    0 => "",
69                    1 => "0",
70                    2 => "00",
71                    3 => "000",
72                    _ => unreachable!(),
73                }
74            }
75
76            const fn name(&self) -> &'static str {
77                match self {$(
78                    Self::$error_name $({ $($field: _),+ })? => concat!(stringify!($name), "::", stringify!($error_name)),
79                )*}
80            }
81        }
82
83        impl fmt::Display for $name {
84            fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
85                write!(
86                    f,
87                    concat!("[", $code_pfx, "{}{}] ", $message_pfx, ": {}"),
88                    self.padding(),
89                    self.code(),
90                    self.message()
91                )
92            }
93        }
94
95        impl fmt::Debug for $name {
96            fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
97                let mut debug_struct = f.debug_struct(self.name());
98                debug_struct.field("message", &format!("{}", self));
99                $(
100                    $(
101                        if let Self::$error_name { $($field),+ } = &self {
102                            $(debug_struct.field(stringify!($field), &$field);)+
103                        }
104                    )?
105                )*
106                debug_struct.finish()
107            }
108        }
109
110        impl std::error::Error for $name {
111            fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
112                None
113            }
114        }
115    };
116}
117
118error_messages! { AnalyzeError
119    code: "ANZ", type: "Analyze Error",
120    MissingResponseField { field: &'static str } =
121        1: "Missing field in message received from server: '{field}'. This is either a version compatibility issue or a bug.",
122    UnknownEnumValue { enum_name: &'static str, value: i32 } =
123        2: "Value '{value}' is out of bounds for enum '{enum_name}'. This is either a version compatibility issue or a bug.",
124}
125
126error_messages! { ConnectionError
127    code: "CXN", type: "Connection Error",
128    RPCMethodUnavailable { message: String } =
129        1: "The server does not support this method, please check the driver-server compatibility:\n'{message}'.",
130    ServerConnectionFailed { addresses: Vec<Address> } =
131        2: "Unable to connect to TypeDB server(s) at: \n{addresses:?}",
132    ServerConnectionFailedWithError { error: String } =
133        3: "Unable to connect to TypeDB server(s), received errors: \n{error}",
134    ServerConnectionFailedStatusError { error: String } =
135        4: "Unable to connect to TypeDB server(s), received network or protocol error: \n{error}",
136    ServerConnectionIsClosed =
137        5: "The connection has been closed and no further operation is allowed.",
138    TransactionIsClosed =
139        6: "The transaction is closed and no further operation is allowed.",
140    TransactionIsClosedWithErrors { errors: String } =
141        7: "The transaction is closed because of the error(s):\n{errors}",
142    DatabaseNotFound { name: String } =
143        8: "Database '{name}' not found.",
144    MissingResponseField { field: &'static str } =
145        9: "Missing field in message received from server: '{field}'. This is either a version compatibility issue or a bug.",
146    UnknownRequestId { request_id: RequestID } =
147        10: "Received a response with unknown request id '{request_id}'",
148    UnexpectedResponse { response: String } =
149        11: "Received unexpected response from server: '{response}'. This is either a version compatibility issue or a bug.",
150    InvalidResponseField { name: &'static str } =
151        12: "Invalid field in message received from server: '{name}'. This is either a version compatibility issue or a bug.",
152    QueryStreamNoResponse =
153        13: "Didn't receive any server responses for the query.",
154    UnexpectedQueryType { query_type: i32 } =
155        14: "Unexpected query type in message received from server: {query_type}. This is either a version compatibility issue or a bug.",
156    ClusterReplicaNotPrimary =
157        15: "The replica is not the primary replica.",
158    ClusterAllNodesFailed { errors: String } =
159        16: "Attempted connecting to all TypeDB Cluster servers, but the following errors occurred: \n{errors}.",
160    TokenCredentialInvalid =
161        17: "Invalid token credentials.",
162    EncryptionSettingsMismatch =
163        18: "Unable to connect to TypeDB: possible encryption settings mismatch.",
164    SSLCertificateNotValidated =
165        19: "SSL handshake with TypeDB failed: the server's identity could not be verified. Possible CA mismatch.",
166    BrokenPipe =
167        20: "Stream closed because of a broken pipe. This could happen if you are attempting to connect to an unencrypted TypeDB server using a TLS-enabled credentials.",
168    ConnectionFailed =
169        21: "Connection failed. Please check the server is running and the address is accessible. Encrypted TypeDB endpoints may also have misconfigured SSL certificates.",
170    MissingPort { address: String } =
171        22: "Invalid URL '{address}': missing port.",
172    AddressTranslationMismatch { unknown: HashSet<Address>, unmapped: HashSet<Address> } =
173        23: "Address translation map does not match the server's advertised address list. User-provided servers not in the advertised list: {unknown:?}. Advertised servers not mapped by user: {unmapped:?}.",
174    ValueTimeZoneNameNotRecognised { time_zone: String } =
175        24: "Time zone provided by the server has name '{time_zone}', which is not an officially recognized timezone.",
176    ValueTimeZoneOffsetNotRecognised { offset: i32 } =
177        25: "Time zone provided by the server has numerical offset '{offset}', which is not recognised as a valid value for offset in seconds.",
178    ValueStructNotImplemented =
179        26: "Struct valued responses are not yet supported by the driver.",
180    ListsNotImplemented =
181        27: "Lists are not yet supported by the driver.",
182    UnexpectedKind { kind: i32 } =
183        28: "Unexpected kind in message received from server: {kind}. This is either a version compatibility issue or a bug.",
184    UnexpectedConnectionClose =
185        29: "Connection closed unexpectedly.",
186    DatabaseImportChannelIsClosed =
187        30: "The database import channel is closed and no further operation is allowed.",
188    DatabaseImportStreamUnexpectedResponse =
189        31: "The database import stream received an unexpected response in the process. It is either a version compatibility issue or a bug.",
190    DatabaseExportChannelIsClosed =
191        32: "The database export channel is closed and no further operation is allowed.",
192    DatabaseExportStreamNoResponse =
193        33: "Didn't receive any server responses for the database export command.",
194    AbsentTlsConfigForTlsConnection =
195        34: "Could not establish a TLS connection without a TLS config specified. Please verify your driver options.",
196    TlsConnectionWithoutHttps =
197        35: "TLS connections can only be enabled when connecting to HTTPS endpoints, for example using 'https://<ip>:port'. Please modify the address, or disable TLS (WARNING: this will send passwords over plaintext).",
198    NonTlsConnectionWithHttps =
199        36: "Connecting to an https endpoint requires enabling TLS in driver options.",
200    AnalyzeNoResponse =
201        37: "Didn't receive any server responses for the analyze request.",
202}
203
204error_messages! { ConceptError
205    code: "CPT", type: "Concept Error",
206    UnavailableRowVariable { variable: String } =
207        1: "Cannot get concept from a concept row by variable '{variable}'.",
208    UnavailableRowIndex { index: usize } =
209        2: "Cannot get concept from a concept row by index '{index}'.",
210}
211
212error_messages! { MigrationError
213    code: "MGT", type: "Migration Error",
214    CannotOpenImportFile { path: String, reason: String } =
215        1: "Cannot open import file '{path}': {reason}",
216    CannotCreateExportFile { path: String, reason: String } =
217        2: "Cannot create export file '{path}': {reason}",
218    CannotExportToTheSameFile =
219        3: "Cannot export both schema and data to the same file.",
220    CannotDecodeImportedConcept =
221        4: "Cannot decode a concept from the provided import file. Make sure to pass a correct database file produced by a TypeDB export operation.",
222    CannotDecodeImportedConceptLength =
223        5: "Cannot decode a concept length from the provided import file. Make sure to pass a correct database file produced by a TypeDB export operation.",
224    CannotEncodeExportedConcept =
225        6: "Cannot encode a concept for export. It's either a version compatibility error or a bug."
226}
227
228error_messages! { InternalError
229    code: "INT", type: "Internal Error",
230    RecvError =
231        1: "Channel is closed.",
232    SendError =
233        2: "Unable to send response over callback channel (receiver dropped).",
234    UnexpectedRequestType { request_type: String } =
235        3: "Unexpected request type for remote procedure call: {request_type}. This is either a version compatibility issue or a bug.",
236    UnexpectedResponseType { response_type: String } =
237        4: "Unexpected response type for remote procedure call: {response_type}. This is either a version compatibility issue or a bug.",
238    UnknownServer { server: Address } =
239        5: "Received replica at unrecognized server: {server}.",
240    EnumOutOfBounds { value: i32, enum_name: &'static str } =
241        6: "Value '{value}' is out of bounds for enum '{enum_name}'.",
242}
243
244#[derive(Clone, PartialEq, Eq)]
245pub struct ServerError {
246    error_code: String,
247    error_domain: String,
248    message: String,
249    stack_trace: Vec<String>,
250}
251
252impl ServerError {
253    pub(crate) fn new(error_code: String, error_domain: String, message: String, stack_trace: Vec<String>) -> Self {
254        Self { error_code, error_domain, message, stack_trace }
255    }
256
257    pub(crate) fn format_code(&self) -> &str {
258        &self.error_code
259    }
260
261    pub(crate) fn message(&self) -> String {
262        self.to_string()
263    }
264}
265
266impl fmt::Display for ServerError {
267    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
268        if self.stack_trace.is_empty() {
269            write!(f, "[{}] {}. {}", self.error_code, self.error_domain, self.message)
270        } else {
271            write!(f, "\n{}", self.stack_trace.join("\nCaused: "))
272        }
273    }
274}
275
276impl fmt::Debug for ServerError {
277    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
278        fmt::Display::fmt(self, f)
279    }
280}
281
282/// Represents errors encountered during operation.
283#[derive(Clone, Debug, PartialEq, Eq)]
284pub enum Error {
285    Analyze(AnalyzeError),
286    Connection(ConnectionError),
287    Concept(ConceptError),
288    Migration(MigrationError),
289    Internal(InternalError),
290    Server(ServerError),
291    Other(String),
292}
293
294impl Error {
295    pub fn code(&self) -> String {
296        match self {
297            Self::Analyze(error) => error.format_code(),
298            Self::Connection(error) => error.format_code(),
299            Self::Concept(error) => error.format_code(),
300            Self::Migration(error) => error.format_code(),
301            Self::Internal(error) => error.format_code(),
302            Self::Server(error) => error.format_code().to_owned(),
303            Self::Other(_error) => String::new(),
304        }
305    }
306
307    pub fn message(&self) -> String {
308        match self {
309            Self::Analyze(error) => error.message(),
310            Self::Connection(error) => error.message(),
311            Self::Concept(error) => error.message(),
312            Self::Migration(error) => error.message(),
313            Self::Internal(error) => error.message(),
314            Self::Server(error) => error.message(),
315            Self::Other(error) => error.clone(),
316        }
317    }
318
319    fn try_extracting_connection_error(_status: &Status, code: &str) -> Option<ConnectionError> {
320        // TODO: We should probably catch more connection errors instead of wrapping them into
321        // ServerErrors. However, the most valuable information even for connection is inside
322        // stacktraces now.
323        match code {
324            "AUT2" | "AUT3" => Some(ConnectionError::TokenCredentialInvalid {}),
325            _ => None,
326        }
327    }
328
329    fn from_message(message: String) -> Self {
330        // TODO: Consider converting some of the messages to connection errors
331        Self::Other(message)
332    }
333
334    fn parse_unavailable(status_message: &str) -> Error {
335        if status_message == "broken pipe" {
336            Error::Connection(ConnectionError::BrokenPipe)
337        } else if status_message.contains("received corrupt message") {
338            Error::Connection(ConnectionError::EncryptionSettingsMismatch)
339        } else if status_message.contains("UnknownIssuer") {
340            Error::Connection(ConnectionError::SSLCertificateNotValidated)
341        } else if status_message.contains("Connection refused") {
342            Error::Connection(ConnectionError::ConnectionFailed)
343        } else {
344            Error::Connection(ConnectionError::ServerConnectionFailedStatusError { error: status_message.to_owned() })
345        }
346    }
347}
348
349impl fmt::Display for Error {
350    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
351        match self {
352            Self::Analyze(error) => write!(f, "{error}"),
353            Self::Connection(error) => write!(f, "{error}"),
354            Self::Concept(error) => write!(f, "{error}"),
355            Self::Migration(error) => write!(f, "{error}"),
356            Self::Internal(error) => write!(f, "{error}"),
357            Self::Server(error) => write!(f, "{error}"),
358            Self::Other(message) => write!(f, "{message}"),
359        }
360    }
361}
362
363impl StdError for Error {
364    fn source(&self) -> Option<&(dyn StdError + 'static)> {
365        match self {
366            Self::Analyze(error) => Some(error),
367            Self::Connection(error) => Some(error),
368            Self::Concept(error) => Some(error),
369            Self::Migration(error) => Some(error),
370            Self::Internal(error) => Some(error),
371            Self::Server(_) => None,
372            Self::Other(_) => None,
373        }
374    }
375}
376
377impl From<AnalyzeError> for Error {
378    fn from(error: AnalyzeError) -> Self {
379        Self::Analyze(error)
380    }
381}
382
383impl From<ConnectionError> for Error {
384    fn from(error: ConnectionError) -> Self {
385        Self::Connection(error)
386    }
387}
388
389impl From<ConceptError> for Error {
390    fn from(error: ConceptError) -> Self {
391        Self::Concept(error)
392    }
393}
394
395impl From<InternalError> for Error {
396    fn from(error: InternalError) -> Self {
397        Self::Internal(error)
398    }
399}
400
401impl From<ServerError> for Error {
402    fn from(error: ServerError) -> Self {
403        Self::Server(error)
404    }
405}
406
407impl From<Status> for Error {
408    fn from(status: Status) -> Self {
409        if let Ok(details) = status.check_error_details() {
410            if let Some(bad_request) = details.bad_request() {
411                Self::Connection(ConnectionError::ServerConnectionFailedWithError {
412                    error: format!("{:?}", bad_request),
413                })
414            } else if let Some(error_info) = details.error_info() {
415                let code = error_info.reason.clone();
416                if let Some(connection_error) = Self::try_extracting_connection_error(&status, &code) {
417                    return Self::Connection(connection_error);
418                }
419                let domain = error_info.domain.clone();
420                let stack_trace =
421                    if let Some(debug_info) = details.debug_info() { debug_info.stack_entries.clone() } else { vec![] };
422
423                Self::Server(ServerError::new(code, domain, status.message().to_owned(), stack_trace))
424            } else {
425                Self::from_message(concat_source_messages(&status))
426            }
427        } else {
428            if status.code() == Code::Unavailable {
429                Self::parse_unavailable(status.message())
430            } else if status.code() == Code::Unknown
431                || is_rst_stream(&status)
432                || status.code() == Code::FailedPrecondition
433                || status.code() == Code::AlreadyExists
434            {
435                Self::Connection(ConnectionError::ServerConnectionFailedStatusError {
436                    error: status.message().to_owned(),
437                })
438            } else if status.code() == Code::Unimplemented {
439                Self::Connection(ConnectionError::RPCMethodUnavailable { message: status.message().to_owned() })
440            } else {
441                Self::from_message(concat_source_messages(&status))
442            }
443        }
444    }
445}
446
447fn is_rst_stream(status: &Status) -> bool {
448    // "Received Rst Stream" occurs if the server is in the process of shutting down.
449    status.message().contains("Received Rst Stream")
450}
451
452fn concat_source_messages(status: &Status) -> String {
453    let mut errors = String::new();
454    errors.push_str(status.message());
455    let mut err: Option<&dyn std::error::Error> = status.source();
456    while let Some(e) = err {
457        errors.push_str(": ");
458        errors.push_str(e.to_string().as_str());
459        err = e.source();
460    }
461    errors
462}
463
464impl From<http::uri::InvalidUri> for Error {
465    fn from(err: http::uri::InvalidUri) -> Self {
466        Self::Other(err.to_string())
467    }
468}
469
470impl From<tonic::transport::Error> for Error {
471    fn from(err: tonic::transport::Error) -> Self {
472        Self::Other(err.to_string())
473    }
474}
475
476impl<T> From<tokio::sync::mpsc::error::SendError<T>> for Error {
477    fn from(_err: tokio::sync::mpsc::error::SendError<T>) -> Self {
478        Self::Internal(InternalError::SendError)
479    }
480}
481
482impl From<tokio::sync::oneshot::error::RecvError> for Error {
483    fn from(_err: tokio::sync::oneshot::error::RecvError) -> Self {
484        Self::Internal(InternalError::RecvError)
485    }
486}
487
488impl From<crossbeam::channel::RecvError> for Error {
489    fn from(_err: crossbeam::channel::RecvError) -> Self {
490        Self::Internal(InternalError::RecvError)
491    }
492}
493
494impl<T> From<crossbeam::channel::SendError<T>> for Error {
495    fn from(_err: crossbeam::channel::SendError<T>) -> Self {
496        Self::Internal(InternalError::SendError)
497    }
498}
499
500impl From<String> for Error {
501    fn from(err: String) -> Self {
502        Self::Other(err)
503    }
504}
505
506impl From<std::io::Error> for Error {
507    fn from(err: std::io::Error) -> Self {
508        Self::Other(err.to_string())
509    }
510}
typedb_driver/common/error.rs

typedb_driver/common/
error.rs
