1use std::any::Any;
16use std::sync::Arc;
17
18use common_error::ext::{ErrorExt, RetryHint, retry_hint_from_io_error};
19use common_error::status_code::StatusCode;
20use common_macro::stack_trace_debug;
21use common_runtime::error::Error as RuntimeError;
22use common_wal::kafka::rskafka_client_error_to_retry_hint;
23use object_store::error::retry_hint_from_opendal_error;
24use serde_json::error::Error as JsonError;
25use snafu::{Location, Snafu};
26use store_api::storage::RegionId;
27
28#[derive(Snafu)]
29#[snafu(visibility(pub))]
30#[stack_trace_debug]
31pub enum Error {
32 #[snafu(display("Failed to create TLS Config"))]
33 TlsConfig {
34 #[snafu(implicit)]
35 location: Location,
36 source: common_wal::error::Error,
37 },
38
39 #[snafu(display("Invalid provider type, expected: {}, actual: {}", expected, actual))]
40 InvalidProvider {
41 #[snafu(implicit)]
42 location: Location,
43 expected: String,
44 actual: String,
45 },
46
47 #[snafu(display("Failed to start log store task: {}", name))]
48 StartWalTask {
49 name: String,
50 #[snafu(implicit)]
51 location: Location,
52 source: RuntimeError,
53 },
54
55 #[snafu(display("Failed to stop log store task: {}", name))]
56 StopWalTask {
57 name: String,
58 #[snafu(implicit)]
59 location: Location,
60 source: RuntimeError,
61 },
62
63 #[snafu(display("Failed to add entry to LogBatch"))]
64 AddEntryLogBatch {
65 #[snafu(source)]
66 error: raft_engine::Error,
67 #[snafu(implicit)]
68 location: Location,
69 },
70
71 #[snafu(display("Failed to perform raft-engine operation"))]
72 RaftEngine {
73 #[snafu(source)]
74 error: raft_engine::Error,
75 #[snafu(implicit)]
76 location: Location,
77 },
78
79 #[snafu(display("Failed to perform IO on path: {}", path))]
80 Io {
81 path: String,
82 #[snafu(source)]
83 error: std::io::Error,
84 #[snafu(implicit)]
85 location: Location,
86 },
87
88 #[snafu(display("Log store not started yet"))]
89 IllegalState {
90 #[snafu(implicit)]
91 location: Location,
92 },
93
94 #[snafu(display("Namespace is illegal: {}", ns))]
95 IllegalNamespace {
96 ns: u64,
97 #[snafu(implicit)]
98 location: Location,
99 },
100
101 #[snafu(display(
102 "Failed to fetch entries from namespace: {}, start: {}, end: {}, max size: {}",
103 ns,
104 start,
105 end,
106 max_size,
107 ))]
108 FetchEntry {
109 ns: u64,
110 start: u64,
111 end: u64,
112 max_size: usize,
113 #[snafu(source)]
114 error: raft_engine::Error,
115 #[snafu(implicit)]
116 location: Location,
117 },
118
119 #[snafu(display(
120 "Cannot override compacted entry, namespace: {}, first index: {}, attempt index: {}",
121 namespace,
122 first_index,
123 attempt_index
124 ))]
125 OverrideCompactedEntry {
126 namespace: u64,
127 first_index: u64,
128 attempt_index: u64,
129 #[snafu(implicit)]
130 location: Location,
131 },
132
133 #[snafu(display(
134 "Failed to build a Kafka client, broker endpoints: {:?}",
135 broker_endpoints
136 ))]
137 BuildClient {
138 broker_endpoints: Vec<String>,
139 #[snafu(implicit)]
140 location: Location,
141 #[snafu(source)]
142 error: rskafka::client::error::Error,
143 },
144
145 #[snafu(display(
146 "Failed to build a Kafka partition client, topic: {}, partition: {}",
147 topic,
148 partition
149 ))]
150 BuildPartitionClient {
151 topic: String,
152 partition: i32,
153 #[snafu(implicit)]
154 location: Location,
155 #[snafu(source)]
156 error: rskafka::client::error::Error,
157 },
158
159 #[snafu(display("Missing required key in a record"))]
160 MissingKey {
161 #[snafu(implicit)]
162 location: Location,
163 },
164
165 #[snafu(display("Missing required value in a record"))]
166 MissingValue {
167 #[snafu(implicit)]
168 location: Location,
169 },
170
171 #[snafu(display("Failed to produce records to Kafka, topic: {}, size: {}", topic, size))]
172 ProduceRecord {
173 topic: String,
174 size: usize,
175 #[snafu(implicit)]
176 location: Location,
177 #[snafu(source)]
178 error: rskafka::client::producer::Error,
179 },
180
181 #[snafu(display("Failed to produce batch records to Kafka"))]
182 BatchProduce {
183 #[snafu(implicit)]
184 location: Location,
185 #[snafu(source)]
186 error: rskafka::client::error::Error,
187 },
188
189 #[snafu(display("Failed to read a record from Kafka, topic: {}", topic))]
190 ConsumeRecord {
191 topic: String,
192 #[snafu(implicit)]
193 location: Location,
194 #[snafu(source)]
195 error: rskafka::client::error::Error,
196 },
197
198 #[snafu(display("Failed to get the latest offset, topic: {}", topic))]
199 GetOffset {
200 topic: String,
201 #[snafu(implicit)]
202 location: Location,
203 #[snafu(source)]
204 error: rskafka::client::error::Error,
205 },
206
207 #[snafu(display("Failed to do a cast"))]
208 Cast {
209 #[snafu(implicit)]
210 location: Location,
211 },
212
213 #[snafu(display("Failed to encode object into json"))]
214 EncodeJson {
215 #[snafu(implicit)]
216 location: Location,
217 #[snafu(source)]
218 error: JsonError,
219 },
220
221 #[snafu(display("Failed to decode object from json"))]
222 DecodeJson {
223 #[snafu(implicit)]
224 location: Location,
225 #[snafu(source)]
226 error: JsonError,
227 },
228
229 #[snafu(display("The record sequence is not legal, error: {}", error))]
230 IllegalSequence {
231 #[snafu(implicit)]
232 location: Location,
233 error: String,
234 },
235
236 #[snafu(display(
237 "Attempt to append discontinuous log entry, region: {}, last index: {}, attempt index: {}",
238 region_id,
239 last_index,
240 attempt_index
241 ))]
242 DiscontinuousLogIndex {
243 region_id: RegionId,
244 last_index: u64,
245 attempt_index: u64,
246 },
247
248 #[snafu(display("OrderedBatchProducer is stopped",))]
249 OrderedBatchProducerStopped {
250 #[snafu(implicit)]
251 location: Location,
252 },
253
254 #[snafu(display("Failed to wait for ProduceResultReceiver"))]
255 WaitProduceResultReceiver {
256 #[snafu(implicit)]
257 location: Location,
258 #[snafu(source)]
259 error: tokio::sync::oneshot::error::RecvError,
260 },
261
262 #[snafu(display(
263 "The length of meta if exceeded the limit: {}, actual: {}",
264 limit,
265 actual
266 ))]
267 MetaLengthExceededLimit {
268 #[snafu(implicit)]
269 location: Location,
270 limit: usize,
271 actual: usize,
272 },
273
274 #[snafu(display("No max value"))]
275 NoMaxValue {
276 #[snafu(implicit)]
277 location: Location,
278 },
279
280 #[snafu(display("Corrupted WAL object, {}", reason))]
281 CorruptedWalObject {
282 reason: String,
283 #[snafu(implicit)]
284 location: Location,
285 },
286
287 #[snafu(display("Invalid WAL object store, {}", reason))]
288 InvalidWalObjectStore {
289 reason: String,
290 #[snafu(implicit)]
291 location: Location,
292 },
293
294 #[snafu(display(
295 "Object store WAL region mismatch, supplied region: {}, {}",
296 region_id,
297 reason
298 ))]
299 MismatchedWalRegion {
300 region_id: RegionId,
301 reason: String,
302 #[snafu(implicit)]
303 location: Location,
304 },
305
306 #[snafu(display(
307 "Invalid WAL entry range, region: {}, start: {}, end: {}",
308 region_id,
309 start_entry_id,
310 end_entry_id
311 ))]
312 InvalidWalEntryRange {
313 region_id: RegionId,
314 start_entry_id: u64,
315 end_entry_id: u64,
316 #[snafu(implicit)]
317 location: Location,
318 },
319
320 #[snafu(display("Incomplete multipart WAL entry of region {}", region_id))]
321 IncompleteWalEntry {
322 region_id: RegionId,
323 #[snafu(implicit)]
324 location: Location,
325 },
326
327 #[snafu(display("WAL object sequence is exhausted, last sequence: {}", last_object_seq))]
328 WalObjectSequenceExhausted {
329 last_object_seq: u64,
330 #[snafu(implicit)]
331 location: Location,
332 },
333
334 #[snafu(display(
335 "WAL object sequence {} is not settled: the open batch has assigned entry ids under it",
336 object_seq
337 ))]
338 WalObjectSequenceUnsettled {
339 object_seq: u64,
340 #[snafu(implicit)]
341 location: Location,
342 },
343
344 #[snafu(display(
345 "WAL entry positions of region {} in one object are exhausted",
346 region_id
347 ))]
348 WalEntryPositionExhausted {
349 region_id: RegionId,
350 #[snafu(implicit)]
351 location: Location,
352 },
353
354 #[snafu(display("WAL object already exists with different content, path: {}", path))]
355 WalObjectConflict {
356 path: String,
357 #[snafu(implicit)]
358 location: Location,
359 },
360
361 #[snafu(display(
362 "WAL object {} was written by the earlier epoch {}, this store writes epoch {}",
363 path,
364 existing_epoch,
365 epoch
366 ))]
367 StaleWalObject {
368 path: String,
369 existing_epoch: u64,
370 epoch: u64,
371 #[snafu(implicit)]
372 location: Location,
373 },
374
375 #[snafu(display(
376 "WAL object {} that starts epoch {} was already present, so the open cannot tell whether it wrote it",
377 path,
378 epoch
379 ))]
380 UnconfirmedWalEpochStart {
381 path: String,
382 epoch: u64,
383 #[snafu(implicit)]
384 location: Location,
385 },
386
387 #[snafu(display("Failed to {} WAL object, path: {}", operation, path))]
388 WalObjectStore {
389 operation: &'static str,
390 path: String,
391 #[snafu(source)]
392 error: object_store::Error,
393 #[snafu(implicit)]
394 location: Location,
395 },
396
397 #[snafu(display("Failed to fetch a segment of WAL object, path: {}", path))]
398 WalSegmentFetch {
399 path: String,
400 #[snafu(source)]
401 error: tokio::task::JoinError,
402 #[snafu(implicit)]
403 location: Location,
404 },
405
406 #[snafu(display("Invalid WAL object, path: {}", path))]
407 InvalidWalObject {
408 path: String,
409 #[snafu(source(from(Error, Box::new)))]
410 source: Box<Error>,
411 #[snafu(implicit)]
412 location: Location,
413 },
414
415 #[snafu(display(
416 "Object store WAL prefix mismatch, expected: {}, actual: {}",
417 expected,
418 actual
419 ))]
420 MismatchedWalPrefix {
421 expected: String,
422 actual: String,
423 #[snafu(implicit)]
424 location: Location,
425 },
426
427 #[snafu(display(
428 "Object store WAL prefix {} on {} does not hold the history of region {}: {}. \
429 If the WAL was moved, stop the instance, copy every object under the prefix from the \
430 old location to this one and start again. If the old WAL is lost, start once with \
431 `[wal].accept_missing_history = true`, which opens the region without its writes that \
432 were not flushed, and turn the option off again once the instance has written to the WAL",
433 prefix,
434 storage,
435 region_id,
436 reason
437 ))]
438 MissingWalHistory {
439 region_id: RegionId,
440 prefix: String,
441 storage: String,
443 reason: String,
444 #[snafu(implicit)]
445 location: Location,
446 },
447
448 #[snafu(display(
449 "Object store WAL gave up its start object {} after refusing a region without its history, and accepts nothing more until it opens again",
450 path
451 ))]
452 WalStartWithdrawn {
453 path: String,
454 #[snafu(implicit)]
455 location: Location,
456 },
457
458 #[snafu(display("Object store WAL log store is stopped"))]
459 ObjectStoreWalStopped {
460 #[snafu(implicit)]
461 location: Location,
462 },
463
464 #[snafu(display("Object store WAL operation failed"))]
465 ObjectStoreWal {
466 source: Arc<Error>,
467 #[snafu(implicit)]
468 location: Location,
469 },
470
471 #[snafu(display(
472 "Object store WAL rejects appends until the object store accepts its acknowledged backlog"
473 ))]
474 ObjectStoreWalFailing {
475 source: Arc<Error>,
476 #[snafu(implicit)]
477 location: Location,
478 },
479}
480
481pub type Result<T> = std::result::Result<T, Error>;
482
483fn rskafka_client_error_to_status_code(error: &rskafka::client::error::Error) -> StatusCode {
484 match error {
485 rskafka::client::error::Error::Connection(_)
486 | rskafka::client::error::Error::Request(_)
487 | rskafka::client::error::Error::InvalidResponse(_)
488 | rskafka::client::error::Error::ServerError { .. }
489 | rskafka::client::error::Error::RetryFailed(_) => StatusCode::Internal,
490 rskafka::client::error::Error::Timeout => StatusCode::StorageUnavailable,
491 _ => StatusCode::Internal,
492 }
493}
494
495impl ErrorExt for Error {
496 fn as_any(&self) -> &dyn Any {
497 self
498 }
499
500 fn status_code(&self) -> StatusCode {
501 use Error::*;
502
503 match self {
504 TlsConfig { .. }
505 | InvalidProvider { .. }
506 | IllegalNamespace { .. }
507 | MissingKey { .. }
508 | MissingValue { .. }
509 | OverrideCompactedEntry { .. }
510 | InvalidWalObjectStore { .. }
511 | MismatchedWalPrefix { .. }
512 | MissingWalHistory { .. }
513 | MismatchedWalRegion { .. }
514 | IncompleteWalEntry { .. }
515 | InvalidWalEntryRange { .. } => StatusCode::InvalidArguments,
516 StartWalTask { .. }
517 | StopWalTask { .. }
518 | IllegalState { .. }
519 | NoMaxValue { .. }
520 | Cast { .. }
521 | EncodeJson { .. }
522 | DecodeJson { .. }
523 | IllegalSequence { .. }
524 | DiscontinuousLogIndex { .. }
525 | OrderedBatchProducerStopped { .. }
526 | WaitProduceResultReceiver { .. }
527 | MetaLengthExceededLimit { .. }
528 | ObjectStoreWalStopped { .. }
529 | WalStartWithdrawn { .. }
530 | WalSegmentFetch { .. } => StatusCode::Internal,
531
532 CorruptedWalObject { .. }
533 | WalObjectConflict { .. }
534 | WalObjectSequenceExhausted { .. }
535 | WalEntryPositionExhausted { .. } => StatusCode::Unexpected,
536 WalObjectSequenceUnsettled { .. } => StatusCode::IllegalState,
537
538 InvalidWalObject { source, .. } => source.status_code(),
539 ObjectStoreWal { source, .. } => source.status_code(),
540 ObjectStoreWalFailing { .. } => StatusCode::StorageUnavailable,
541
542 WalObjectStore { .. }
544 | StaleWalObject { .. }
545 | UnconfirmedWalEpochStart { .. }
546 | Io { .. } => StatusCode::StorageUnavailable,
547 FetchEntry { .. } | RaftEngine { .. } | AddEntryLogBatch { .. } => {
549 StatusCode::StorageUnavailable
550 }
551 ProduceRecord { error, .. } => match error {
553 rskafka::client::producer::Error::Client(error) => {
554 rskafka_client_error_to_status_code(error)
555 }
556 rskafka::client::producer::Error::Aggregator(_)
557 | rskafka::client::producer::Error::FlushError(_)
558 | rskafka::client::producer::Error::TooLarge => StatusCode::Internal,
559 },
560 BuildClient { error, .. }
561 | BuildPartitionClient { error, .. }
562 | BatchProduce { error, .. }
563 | GetOffset { error, .. }
564 | ConsumeRecord { error, .. } => rskafka_client_error_to_status_code(error),
565 }
566 }
567
568 fn retry_hint(&self) -> RetryHint {
569 use Error::*;
570
571 match self {
572 WalObjectStore { error, .. } => retry_hint_from_opendal_error(error),
573 ObjectStoreWal { source, .. } => source.retry_hint(),
574 Io { error, .. } => retry_hint_from_io_error(error),
575 FetchEntry { .. }
576 | RaftEngine { .. }
577 | AddEntryLogBatch { .. }
578 | WalObjectSequenceUnsettled { .. }
579 | StaleWalObject { .. }
580 | UnconfirmedWalEpochStart { .. }
581 | ObjectStoreWalFailing { .. } => RetryHint::Retryable,
582 ProduceRecord { error, .. } => match error {
583 rskafka::client::producer::Error::Client(error) => {
584 rskafka_client_error_to_retry_hint(error)
585 }
586 rskafka::client::producer::Error::Aggregator(_)
587 | rskafka::client::producer::Error::FlushError(_)
588 | rskafka::client::producer::Error::TooLarge => RetryHint::NonRetryable,
589 },
590 BuildClient { error, .. }
591 | BuildPartitionClient { error, .. }
592 | BatchProduce { error, .. }
593 | GetOffset { error, .. }
594 | ConsumeRecord { error, .. } => rskafka_client_error_to_retry_hint(error),
595 _ => RetryHint::NonRetryable,
596 }
597 }
598}