client/
error.rs

1// Copyright 2023 Greptime Team
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 std::any::Any;
16
17use common_error::define_from_tonic_status;
18use common_error::ext::{BoxedError, ErrorExt, RetryHint};
19use common_error::status_code::StatusCode;
20use common_macro::stack_trace_debug;
21use snafu::{Location, OptionExt, Snafu, location};
22use tonic::Code;
23use tonic::metadata::errors::InvalidMetadataValue;
24
25#[derive(Snafu)]
26#[snafu(visibility(pub))]
27#[stack_trace_debug]
28pub enum Error {
29    #[snafu(display(
30        "Region request has {} failed entries; {}",
31        response.failures.len(),
32        response.first_failure().map(|(index, error)| format!("entry {index}: {}", error.output_msg())).unwrap_or_default()
33    ))]
34    PartialRegion {
35        response: Box<api::region::BatchRegionResponse<Error>>,
36    },
37
38    #[snafu(display("Illegal Flight messages, reason: {}", reason))]
39    IllegalFlightMessages {
40        reason: String,
41        #[snafu(implicit)]
42        location: Location,
43    },
44
45    #[snafu(display("Failed to do Flight get, code: {}", tonic_code))]
46    FlightGet {
47        addr: String,
48        tonic_code: Code,
49        source: BoxedError,
50    },
51
52    #[snafu(display(
53        "Failed to receive Flight data from {}, code: {}: {}",
54        addr,
55        tonic_code,
56        message
57    ))]
58    FlightStream {
59        addr: String,
60        tonic_code: Code,
61        message: String,
62        source: BoxedError,
63        #[snafu(implicit)]
64        location: Location,
65    },
66
67    #[snafu(display("Failed to convert FlightData"))]
68    ConvertFlightData {
69        #[snafu(implicit)]
70        location: Location,
71        source: common_grpc::Error,
72    },
73
74    #[snafu(display("Illegal GRPC client state: {}", err_msg))]
75    IllegalGrpcClientState {
76        err_msg: String,
77        #[snafu(implicit)]
78        location: Location,
79    },
80
81    #[snafu(display("Missing required field in protobuf, field: {}", field))]
82    MissingField {
83        field: String,
84        #[snafu(implicit)]
85        location: Location,
86    },
87
88    #[snafu(display("Failed to create gRPC channel, peer address: {}", addr))]
89    CreateChannel {
90        addr: String,
91        #[snafu(implicit)]
92        location: Location,
93        source: common_grpc::error::Error,
94    },
95
96    #[snafu(display("Failed to create Tls channel manager"))]
97    CreateTlsChannel {
98        #[snafu(implicit)]
99        location: Location,
100        source: common_grpc::error::Error,
101    },
102
103    #[snafu(display("Failed to request RegionServer {}, code: {}", addr, code))]
104    RegionServer {
105        addr: String,
106        code: Code,
107        source: BoxedError,
108        #[snafu(implicit)]
109        location: Location,
110    },
111
112    #[snafu(display("Failed to request FlowServer {}, code: {}", addr, code))]
113    FlowServer {
114        addr: String,
115        code: Code,
116        source: BoxedError,
117        #[snafu(implicit)]
118        location: Location,
119    },
120
121    // Server error carried in a protobuf response status.
122    #[snafu(display("{}", msg))]
123    Server {
124        code: StatusCode,
125        msg: String,
126        retry_hint: RetryHint,
127        #[snafu(implicit)]
128        location: Location,
129    },
130
131    #[snafu(display("Illegal Database response: {err_msg}"))]
132    IllegalDatabaseResponse {
133        err_msg: String,
134        #[snafu(implicit)]
135        location: Location,
136    },
137
138    #[snafu(display("Invalid Tonic metadata value"))]
139    InvalidTonicMetadataValue {
140        #[snafu(source)]
141        error: InvalidMetadataValue,
142        #[snafu(implicit)]
143        location: Location,
144    },
145
146    #[snafu(display("Failed to convert Schema"))]
147    ConvertSchema {
148        #[snafu(implicit)]
149        location: Location,
150        source: datatypes::error::Error,
151    },
152
153    #[snafu(display("{}", msg))]
154    Tonic {
155        code: StatusCode,
156        msg: String,
157        tonic_code: Code,
158        retry_hint: RetryHint,
159        #[snafu(implicit)]
160        location: Location,
161    },
162
163    #[snafu(display("External error"))]
164    External {
165        #[snafu(implicit)]
166        location: Location,
167        source: BoxedError,
168    },
169}
170
171pub type Result<T> = std::result::Result<T, Error>;
172
173impl ErrorExt for Error {
174    fn status_code(&self) -> StatusCode {
175        match self {
176            Error::PartialRegion { response } => response
177                .first_failure()
178                .map_or(StatusCode::Internal, |(_, error)| error.status_code()),
179            Error::IllegalFlightMessages { .. }
180            | Error::MissingField { .. }
181            | Error::IllegalDatabaseResponse { .. } => StatusCode::Internal,
182
183            Error::Server { code, .. } | Error::Tonic { code, .. } => *code,
184            Error::FlightGet { source, .. }
185            | Error::FlightStream { source, .. }
186            | Error::RegionServer { source, .. }
187            | Error::FlowServer { source, .. } => source.status_code(),
188            Error::CreateChannel { source, .. }
189            | Error::ConvertFlightData { source, .. }
190            | Error::CreateTlsChannel { source, .. } => source.status_code(),
191            Error::IllegalGrpcClientState { .. } => StatusCode::Unexpected,
192            Error::InvalidTonicMetadataValue { .. } => StatusCode::InvalidArguments,
193            Error::ConvertSchema { source, .. } => source.status_code(),
194            Error::External { source, .. } => source.status_code(),
195        }
196    }
197
198    fn as_any(&self) -> &dyn Any {
199        self
200    }
201
202    fn retry_hint(&self) -> RetryHint {
203        match self {
204            Error::PartialRegion { .. } => RetryHint::Retryable,
205            Error::Tonic { retry_hint, .. } | Error::Server { retry_hint, .. } => *retry_hint,
206            Error::FlightGet { source, .. }
207            | Error::FlightStream { source, .. }
208            | Error::FlowServer { source, .. }
209            | Error::External { source, .. } => source.retry_hint(),
210            Error::RegionServer { code, source, .. } => {
211                let hint = source.retry_hint();
212                // Preserve the transport retry policy previously applied by should_retry().
213                if !hint.is_retryable()
214                    && matches!(
215                        code,
216                        Code::Unavailable | Code::Cancelled | Code::DeadlineExceeded
217                    )
218                {
219                    RetryHint::Retryable
220                } else {
221                    hint
222                }
223            }
224            Error::ConvertFlightData { source, .. }
225            | Error::CreateChannel { source, .. }
226            | Error::CreateTlsChannel { source, .. } => source.retry_hint(),
227            Error::ConvertSchema { source, .. } => source.retry_hint(),
228            _ => RetryHint::NonRetryable,
229        }
230    }
231}
232
233define_from_tonic_status!(Error, Tonic);
234
235impl From<Error> for common_meta::error::Error {
236    fn from(error: Error) -> Self {
237        match error {
238            Error::PartialRegion { response } => Self::PartialRegion {
239                response: Box::new(api::region::BatchRegionResponse {
240                    affected_rows: response.affected_rows,
241                    extensions: response.extensions,
242                    failures: response
243                        .failures
244                        .into_iter()
245                        .map(|(index, error)| (index, error.into()))
246                        .collect(),
247                }),
248            },
249            error => Self::External {
250                source: BoxedError::new(error),
251                location: location!(),
252            },
253        }
254    }
255}
256
257impl Error {
258    /// Returns the gRPC status code if this error is caused by a gRPC request failure.
259    pub fn tonic_code(&self) -> Option<Code> {
260        match self {
261            Self::FlightGet { tonic_code, .. }
262            | Self::FlightStream { tonic_code, .. }
263            | Self::RegionServer {
264                code: tonic_code, ..
265            }
266            | Self::FlowServer {
267                code: tonic_code, ..
268            }
269            | Self::Tonic { tonic_code, .. } => Some(*tonic_code),
270            _ => None,
271        }
272    }
273
274    /// Returns true if the error is a connection error that may be resolved by retrying the request.
275    pub fn is_connection_error(&self) -> bool {
276        matches!(self.tonic_code(), Some(Code::Unavailable))
277    }
278
279    pub fn should_retry(&self) -> bool {
280        self.retry_hint().is_retryable()
281            || self.is_connection_error()
282            || matches!(
283                self.tonic_code(),
284                Some(Code::Cancelled) | Some(Code::DeadlineExceeded)
285            )
286    }
287}
288
289impl Error {
290    /// Decodes an application error carried in a protobuf response.
291    pub(crate) fn try_from_response_status(status: &api::v1::Status) -> Result<Self> {
292        let code = StatusCode::from_u32(status.status_code).with_context(|| {
293            IllegalDatabaseResponseSnafu {
294                err_msg: format!("Unknown server status: {}", status.status_code),
295            }
296        })?;
297        Ok(ServerSnafu {
298            code,
299            msg: status.err_msg.clone(),
300            retry_hint: status.retry_hint.parse().unwrap_or(RetryHint::NonRetryable),
301        }
302        .build())
303    }
304}
client/error.rs

client/
error.rs
