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::{error::Error as StdError, fmt, time::Duration};
21
22use tonic::{Code, Status};
23use tonic_types::StatusExt;
24
25use super::RequestID;
26use crate::common::address::{Address, Addresses};
27
28macro_rules! error_messages {
29    {
30        $name:ident code: $code_pfx:literal, type: $message_pfx:literal,
31        $($error_name:ident $({ $($field:ident : $inner:ty),+ $(,)? })? = $code:literal: $body:literal),+ $(,)?
32    } => {
33        #[derive(Clone, Eq, PartialEq)]
34        pub enum $name {$(
35            $error_name$( { $($field: $inner),+ })?,
36        )*}
37
38        impl $name {
39            pub const PREFIX: &'static str = $code_pfx;
40
41            pub const fn code(&self) -> usize {
42                match self {$(
43                    Self::$error_name $({ $($field: _),+ })? => $code,
44                )*}
45            }
46
47            pub fn format_code(&self) -> String {
48                format!(concat!("[", $code_pfx, "{}{}]"), self.padding(), self.code())
49            }
50
51            pub fn message(&self) -> String {
52                match self {$(
53                    Self::$error_name $({$($field),+})? => format!($body $($(, $field = $field)+)?),
54                )*}
55            }
56
57            const fn max_code() -> usize {
58                let mut max = usize::MIN;
59                $(max = if $code > max { $code } else { max };)*
60                max
61            }
62
63            const fn num_digits(x: usize) -> usize {
64                if (x < 10) { 1 } else { 1 + Self::num_digits(x/10) }
65            }
66
67            const fn padding(&self) -> &'static str {
68                match Self::num_digits(Self::max_code()) - Self::num_digits(self.code()) {
69                    0 => "",
70                    1 => "0",
71                    2 => "00",
72                    3 => "000",
73                    _ => unreachable!(),
74                }
75            }
76
77            const fn name(&self) -> &'static str {
78                match self {$(
79                    Self::$error_name $({ $($field: _),+ })? => concat!(stringify!($name), "::", stringify!($error_name)),
80                )*}
81            }
82        }
83
84        impl fmt::Display for $name {
85            fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
86                write!(
87                    f,
88                    concat!("[", $code_pfx, "{}{}] ", $message_pfx, ": {}"),
89                    self.padding(),
90                    self.code(),
91                    self.message()
92                )
93            }
94        }
95
96        impl fmt::Debug for $name {
97            fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
98                let mut debug_struct = f.debug_struct(self.name());
99                debug_struct.field("message", &format!("{}", self));
100                $(
101                    $(
102                        if let Self::$error_name { $($field),+ } = &self {
103                            $(debug_struct.field(stringify!($field), &$field);)+
104                        }
105                    )?
106                )*
107                debug_struct.finish()
108            }
109        }
110
111        impl std::error::Error for $name {
112            fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
113                None
114            }
115        }
116    };
117}
118
119error_messages! { AnalyzeError
120    code: "ANZ", type: "Analyze Error",
121    MissingResponseField { field: &'static str } =
122        1: "Missing field in message received from server: '{field}'. This is either a version compatibility issue or a bug.",
123    UnknownEnumValue { enum_name: &'static str, value: i32 } =
124        2: "Value '{value}' is out of bounds for enum '{enum_name}'. This is either a version compatibility issue or a bug.",
125}
126
127error_messages! { ConnectionError
128    code: "CXN", type: "Connection Error",
129    RPCMethodUnavailable { message: String } =
130        1: "The server does not support this method, please check the driver-server compatibility:\n'{message}'.",
131    ServerConnectionFailed { configured_addresses: Addresses, accessed_addresses: Addresses, details: String } =
132        2: "Unable to connect to TypeDB server(s).\nInitially configured addresses: {configured_addresses}.\nTried accessing addresses: {accessed_addresses}. Details: {details}",
133    ServerConnectionFailedWithError { error: String } =
134        3: "Unable to connect to TypeDB server(s), received errors: \n{error}",
135    ServerConnectionFailedNetworking { error: String } =
136        4: "Unable to connect to TypeDB server(s), received network or protocol error: \n{error}",
137    ServerConnectionIsClosed =
138        5: "The connection has been closed and no further operation is allowed.",
139    ServerConnectionIsClosedUnexpectedly =
140        6: "The connection has been closed unexpectedly and no further operation is allowed.",
141    TransactionIsClosed =
142        7: "The transaction is closed and no further operation is allowed.",
143    TransactionIsClosedWithErrors { errors: String } =
144        8: "The transaction is closed because of the error(s):\n{errors}",
145    MissingResponseField { field: &'static str } =
146        9: "Missing field in message received from server: '{field}'. This is either a version compatibility issue or a bug.",
147    UnknownRequestId { request_id: RequestID } =
148        10: "Received a response with unknown request id '{request_id}'",
149    UnexpectedResponse { response: String } =
150        11: "Received unexpected response from server: '{response}'. This is either a version compatibility issue or a bug.",
151    QueryStreamNoResponse =
152        12: "Didn't receive any server responses for the query.",
153    UnexpectedQueryType { query_type: i32 } =
154        13: "Unexpected query type in message received from server: {query_type}. This is either a version compatibility issue or a bug.",
155    ClusterServerNotPrimary =
156        14: "The server is not primary.",
157    UnknownDirectServerRouting { address: Address, known_addresses: Addresses } =
158        15: "Could not execute operation against '{address}' since it's not in the list of known servers: {known_addresses}.",
159    TokenCredentialInvalid =
160        16: "Invalid credential supplied.",
161    EncryptionSettingsMismatch =
162        17: "Unable to connect to TypeDB: possible encryption settings mismatch.",
163    SslCertificateNotValidated =
164        18: "SSL handshake with TypeDB failed: the server's identity could not be verified. Possible CA mismatch.",
165    BrokenPipe =
166        19: "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.",
167    ConnectionRefusedNetworking =
168        20: "Connection refused. Please check the server is running and the address is accessible. Encrypted TypeDB endpoints may also have misconfigured SSL certificates.",
169    MissingPort { address: String } =
170        21: "Invalid URL '{address}': missing port.",
171    UnexpectedServerReplicationRole { replication_role: i32 } =
172        22: "Unexpected replication role in message received from server: {replication_role}. This is either a version compatibility issue or a bug.",
173    ValueTimeZoneNameNotRecognised { time_zone: String } =
174        23: "Time zone provided by the server has name '{time_zone}', which is not an officially recognized timezone.",
175    ValueTimeZoneOffsetNotRecognised { offset: i32 } =
176        24: "Time zone provided by the server has numerical offset '{offset}', which is not recognised as a valid value for offset in seconds.",
177    ValueStructNotImplemented =
178        25: "Struct valued responses are not yet supported by the driver.",
179    ListsNotImplemented =
180        26: "Lists are not yet supported by the driver.",
181    UnexpectedKind { kind: i32 } =
182        27: "Unexpected kind in message received from server: {kind}. This is either a version compatibility issue or a bug.",
183    UnexpectedConnectionClose =
184        28: "Connection closed unexpectedly.",
185    DatabaseImportChannelIsClosed =
186        29: "The database import channel is closed and no further operation is allowed.",
187    DatabaseImportStreamUnexpectedResponse =
188        30: "The database import stream received an unexpected response in the process. It is either a version compatibility issue or a bug.",
189    DatabaseExportChannelIsClosed =
190        31: "The database export channel is closed and no further operation is allowed.",
191    DatabaseExportStreamNoResponse =
192        32: "Didn't receive any server responses for the database export command.",
193    AbsentTlsConfigForTlsConnection =
194        33: "Could not establish a TLS connection without a TLS config specified. Please verify your driver options.",
195    ServerIsNotInitialised =
196        34: "Server is not yet initialized.",
197    AnalyzeNoResponse =
198        35: "Didn't receive any server responses for the analyze request.",
199    NotPrimaryOnReadOnly { address: Address } =
200        36: "Could not execute a readonly operation on a non-primary server '{address}'.",
201    NoPrimaryServer =
202        37: "Could not find a primary server.",
203    SchemeTlsSettingsMismatch { scheme: http::uri::Scheme, is_tls_enabled: bool } =
204        38: "Scheme {scheme} is not compatible with tls setting `enabled: {is_tls_enabled}`",
205    RequestTimeout { timeout: String } =
206        39: "Request timed out after {timeout}. The server may be unresponsive.",
207}
208
209impl ConnectionError {
210    pub fn request_timeout(timeout: Duration) -> Self {
211        let (value, decimal, unit) = if timeout.as_secs() > 0 {
212            let decimal = timeout.subsec_millis();
213            (timeout.as_secs(), (decimal > 0).then_some(decimal), "seconds")
214        } else if timeout.subsec_millis() > 0 {
215            let decimal = timeout.subsec_micros() % 1000;
216            (timeout.subsec_millis() as u64, (decimal > 0).then_some(decimal), "milliseconds")
217        } else if timeout.subsec_micros() > 0 {
218            let decimal = timeout.subsec_nanos() % 1000;
219            (timeout.subsec_micros() as u64, (decimal > 0).then_some(decimal), "microseconds")
220        } else {
221            (timeout.subsec_nanos() as u64, None, "nanoseconds")
222        };
223
224        let timeout_str = match decimal {
225            Some(d) => format!("{value}.{d:03} {unit}"),
226            None => format!("{value} {unit}"),
227        };
228        ConnectionError::RequestTimeout { timeout: timeout_str }
229    }
230}
231
232error_messages! { ConceptError
233    code: "CPT", type: "Concept Error",
234    UnavailableRowVariable { variable: String } =
235        1: "Cannot get concept from a concept row by variable '{variable}'.",
236    UnavailableRowIndex { index: usize } =
237        2: "Cannot get concept from a concept row by index '{index}'.",
238}
239
240error_messages! { MigrationError
241    code: "MGT", type: "Migration Error",
242    CannotOpenImportFile { path: String, reason: String } =
243        1: "Cannot open import file '{path}': {reason}",
244    CannotCreateExportFile { path: String, reason: String } =
245        2: "Cannot create export file '{path}': {reason}",
246    CannotExportToTheSameFile =
247        3: "Cannot export both schema and data to the same file.",
248    CannotDecodeImportedConcept =
249        4: "Cannot decode a concept from the provided import file. Make sure to pass a correct database file produced by a TypeDB export operation.",
250    CannotDecodeImportedConceptLength =
251        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.",
252    CannotEncodeExportedConcept =
253        6: "Cannot encode a concept for export. It's either a version compatibility error or a bug."
254}
255
256error_messages! { InternalError
257    code: "INT", type: "Internal Error",
258    RecvError =
259        1: "Channel is closed.",
260    SendError =
261        2: "Unable to send response over callback channel (receiver dropped).",
262    UnexpectedRequestType { request_type: String } =
263        3: "Unexpected request type for remote procedure call: {request_type}. This is either a version compatibility issue or a bug.",
264    UnexpectedResponseType { response_type: String } =
265        4: "Unexpected response type for remote procedure call: {response_type}. This is either a version compatibility issue or a bug.",
266    Unimplemented { details: String } =
267        5: "Unimplemented feature: {details}.",
268}
269
270#[derive(Clone, PartialEq, Eq)]
271pub struct ServerError {
272    error_code: String,
273    error_domain: String,
274    message: String,
275    stack_trace: Vec<String>,
276}
277
278impl ServerError {
279    pub(crate) fn new(error_code: String, error_domain: String, message: String, stack_trace: Vec<String>) -> Self {
280        Self { error_code, error_domain, message, stack_trace }
281    }
282
283    pub(crate) fn format_code(&self) -> &str {
284        &self.error_code
285    }
286
287    pub(crate) fn message(&self) -> String {
288        self.to_string()
289    }
290}
291
292impl fmt::Display for ServerError {
293    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
294        if self.stack_trace.is_empty() {
295            write!(f, "[{}] {}. {}", self.error_code, self.error_domain, self.message)
296        } else {
297            write!(f, "\n{}", self.stack_trace.join("\nCaused: "))
298        }
299    }
300}
301
302impl fmt::Debug for ServerError {
303    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
304        fmt::Display::fmt(self, f)
305    }
306}
307
308/// Represents errors encountered during operation.
309#[derive(Clone, Debug, PartialEq, Eq)]
310pub enum Error {
311    Analyze(AnalyzeError),
312    Connection(ConnectionError),
313    Concept(ConceptError),
314    Migration(MigrationError),
315    Internal(InternalError),
316    Server(ServerError),
317    Other(String),
318}
319
320impl Error {
321    pub fn code(&self) -> String {
322        match self {
323            Self::Analyze(error) => error.format_code(),
324            Self::Connection(error) => error.format_code(),
325            Self::Concept(error) => error.format_code(),
326            Self::Migration(error) => error.format_code(),
327            Self::Internal(error) => error.format_code(),
328            Self::Server(error) => error.format_code().to_owned(),
329            Self::Other(_error) => String::new(),
330        }
331    }
332
333    pub fn message(&self) -> String {
334        match self {
335            Self::Analyze(error) => error.message(),
336            Self::Connection(error) => error.message(),
337            Self::Concept(error) => error.message(),
338            Self::Migration(error) => error.message(),
339            Self::Internal(error) => error.message(),
340            Self::Server(error) => error.message(),
341            Self::Other(error) => error.clone(),
342        }
343    }
344
345    fn try_extracting_connection_error_code(code: &str) -> Option<ConnectionError> {
346        match code {
347            "AUT1" | "AUT2" | "AUT3" => Some(ConnectionError::TokenCredentialInvalid {}),
348            "SRV14" | "RFT1" | "CSV7" => Some(ConnectionError::ServerIsNotInitialised {}),
349            "CSV8" => Some(ConnectionError::ClusterServerNotPrimary {}),
350            _ => None,
351        }
352    }
353
354    fn try_extracting_connection_error_message(message: &str) -> Option<ConnectionError> {
355        if is_rst_stream(message) || is_tcp_connect_error(message) || is_connection_error(message) {
356            Some(ConnectionError::ServerConnectionFailedNetworking { error: message.to_string() })
357        } else if is_reading_body_from_connection_error(message) {
358            Some(ConnectionError::ServerConnectionIsClosedUnexpectedly)
359        } else {
360            None
361        }
362    }
363
364    fn parse_unavailable(status_message: &str) -> Error {
365        if status_message == "broken pipe" {
366            Error::Connection(ConnectionError::BrokenPipe)
367        } else if status_message.contains("received corrupt message") {
368            Error::Connection(ConnectionError::EncryptionSettingsMismatch)
369        } else if status_message.contains("UnknownIssuer") {
370            Error::Connection(ConnectionError::SslCertificateNotValidated)
371        } else if status_message.contains("Connection refused") {
372            Error::Connection(ConnectionError::ConnectionRefusedNetworking)
373        } else {
374            Error::Connection(ConnectionError::ServerConnectionFailedNetworking { error: status_message.to_owned() })
375        }
376    }
377}
378
379impl fmt::Display for Error {
380    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
381        match self {
382            Self::Analyze(error) => write!(f, "{error}"),
383            Self::Connection(error) => write!(f, "{error}"),
384            Self::Concept(error) => write!(f, "{error}"),
385            Self::Migration(error) => write!(f, "{error}"),
386            Self::Internal(error) => write!(f, "{error}"),
387            Self::Server(error) => write!(f, "{error}"),
388            Self::Other(message) => write!(f, "{message}"),
389        }
390    }
391}
392
393impl StdError for Error {
394    fn source(&self) -> Option<&(dyn StdError + 'static)> {
395        match self {
396            Self::Analyze(error) => Some(error),
397            Self::Connection(error) => Some(error),
398            Self::Concept(error) => Some(error),
399            Self::Migration(error) => Some(error),
400            Self::Internal(error) => Some(error),
401            Self::Server(_) => None,
402            Self::Other(_) => None,
403        }
404    }
405}
406
407impl From<AnalyzeError> for Error {
408    fn from(error: AnalyzeError) -> Self {
409        Self::Analyze(error)
410    }
411}
412
413impl From<ConnectionError> for Error {
414    fn from(error: ConnectionError) -> Self {
415        Self::Connection(error)
416    }
417}
418
419impl From<ConceptError> for Error {
420    fn from(error: ConceptError) -> Self {
421        Self::Concept(error)
422    }
423}
424
425impl From<MigrationError> for Error {
426    fn from(error: MigrationError) -> Self {
427        Self::Migration(error)
428    }
429}
430
431impl From<InternalError> for Error {
432    fn from(error: InternalError) -> Self {
433        Self::Internal(error)
434    }
435}
436
437impl From<ServerError> for Error {
438    fn from(error: ServerError) -> Self {
439        Self::Server(error)
440    }
441}
442
443impl From<Status> for Error {
444    fn from(status: Status) -> Self {
445        if let Ok(details) = status.check_error_details() {
446            if let Some(bad_request) = details.bad_request() {
447                return Self::Connection(ConnectionError::ServerConnectionFailedWithError {
448                    error: format!("{:?}", bad_request),
449                });
450            }
451
452            let message = concat_source_messages(&status);
453            if let Some(connection_error) = Self::try_extracting_connection_error_message(&message) {
454                return Self::Connection(connection_error);
455            }
456
457            if let Some(error_info) = details.error_info() {
458                let code = error_info.reason.clone();
459                if let Some(connection_error) = Self::try_extracting_connection_error_code(&code) {
460                    Self::Connection(connection_error)
461                } else {
462                    let domain = error_info.domain.clone();
463                    let stack_trace =
464                        details.debug_info().map(|debug_info| debug_info.stack_entries.clone()).unwrap_or_default();
465                    Self::Server(ServerError::new(code, domain, status.message().to_owned(), stack_trace))
466                }
467            } else {
468                Self::Other(message)
469            }
470        } else {
471            match status.code() {
472                Code::Unavailable => Self::parse_unavailable(status.message()),
473                Code::Unknown | Code::FailedPrecondition | Code::AlreadyExists => {
474                    Self::Connection(ConnectionError::ServerConnectionFailedNetworking {
475                        error: status.message().to_owned(),
476                    })
477                }
478                Code::Unimplemented => {
479                    Self::Connection(ConnectionError::RPCMethodUnavailable { message: status.message().to_owned() })
480                }
481                _ => {
482                    let message = concat_source_messages(&status);
483                    Self::try_extracting_connection_error_message(&message)
484                        .map(|error| Self::Connection(error))
485                        .unwrap_or_else(|| Self::Other(message))
486                }
487            }
488        }
489    }
490}
491
492fn is_rst_stream(message: &str) -> bool {
493    // "Received Rst Stream" occurs if the server is in the process of shutting down.
494    message.contains("Received Rst Stream")
495}
496
497fn is_reading_body_from_connection_error(message: &str) -> bool {
498    // This error can be returned when the server crashes
499    message.contains("error reading a body from connection")
500}
501
502fn is_tcp_connect_error(message: &str) -> bool {
503    // No TCP connection
504    message.contains("tcp connect error")
505}
506
507fn is_connection_error(message: &str) -> bool {
508    // Transport-level connection errors
509    message.contains("connection error")
510}
511
512fn concat_source_messages(status: &Status) -> String {
513    let mut errors = String::new();
514    errors.push_str(status.message());
515    let mut err: Option<&dyn std::error::Error> = status.source();
516    while let Some(e) = err {
517        errors.push_str(": ");
518        errors.push_str(e.to_string().as_str());
519        err = e.source();
520    }
521    errors
522}
523
524impl From<http::uri::InvalidUri> for Error {
525    fn from(err: http::uri::InvalidUri) -> Self {
526        Self::Other(err.to_string())
527    }
528}
529
530impl From<tonic::transport::Error> for Error {
531    fn from(err: tonic::transport::Error) -> Self {
532        Self::Other(err.to_string())
533    }
534}
535
536impl<T> From<tokio::sync::mpsc::error::SendError<T>> for Error {
537    fn from(_err: tokio::sync::mpsc::error::SendError<T>) -> Self {
538        Self::Internal(InternalError::SendError)
539    }
540}
541
542impl From<tokio::sync::oneshot::error::RecvError> for Error {
543    fn from(_err: tokio::sync::oneshot::error::RecvError) -> Self {
544        Self::Internal(InternalError::RecvError)
545    }
546}
547
548impl From<crossbeam::channel::RecvError> for Error {
549    fn from(_err: crossbeam::channel::RecvError) -> Self {
550        Self::Internal(InternalError::RecvError)
551    }
552}
553
554impl<T> From<crossbeam::channel::SendError<T>> for Error {
555    fn from(_err: crossbeam::channel::SendError<T>) -> Self {
556        Self::Internal(InternalError::SendError)
557    }
558}
559
560impl From<String> for Error {
561    fn from(err: String) -> Self {
562        Self::Other(err)
563    }
564}
565
566impl From<std::io::Error> for Error {
567    fn from(err: std::io::Error) -> Self {
568        Self::Other(err.to_string())
569    }
570}
typedb_driver/common/error.rs

typedb_driver/common/
error.rs
