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