reifydb_cdc/
error.rs

1// SPDX-License-Identifier: Apache-2.0
2// Copyright (c) 2026 ReifyDB
3
4use std::{error::Error as StdError, fmt, fmt::Display};
5
6use reifydb_core::common::CommitVersion;
7use reifydb_value::{error, error::Error};
8
9#[derive(Debug, Clone)]
10pub enum CdcError {
11	Internal(String),
12
13	NotFound(CommitVersion),
14
15	Codec(String),
16}
17
18impl Display for CdcError {
19	fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
20		match self {
21			CdcError::Internal(msg) => write!(f, "CDC storage internal error: {}", msg),
22			CdcError::NotFound(version) => write!(f, "CDC entry not found: {:?}", version),
23			CdcError::Codec(msg) => write!(f, "CDC codec error: {}", msg),
24		}
25	}
26}
27
28impl StdError for CdcError {}
29
30impl From<CdcError> for Error {
31	fn from(err: CdcError) -> Self {
32		error!(match err {
33			CdcError::Internal(msg) => diagnostic::storage_error(msg),
34			CdcError::NotFound(version) => diagnostic::not_found(version.0),
35			CdcError::Codec(msg) => diagnostic::codec_error(msg),
36		})
37	}
38}
39
40pub type CdcResult<T> = Result<T, CdcError>;
41
42pub mod diagnostic {
43
44	use reifydb_value::{error::Diagnostic, fragment::Fragment};
45
46	pub fn storage_error(msg: impl Into<String>) -> Diagnostic {
47		Diagnostic {
48			code: "CDC_001".to_string(),
49			rql: None,
50			message: format!("CDC storage error: {}", msg.into()),
51			column: None,
52			fragment: Fragment::None,
53			label: None,
54			help: Some("Check CDC storage configuration and availability".to_string()),
55			notes: vec![],
56			cause: None,
57			operator_chain: None,
58		}
59	}
60
61	pub fn not_found(version: u64) -> Diagnostic {
62		Diagnostic {
63			code: "CDC_002".to_string(),
64			rql: None,
65			message: format!("CDC entry not found for version {}", version),
66			column: None,
67			fragment: Fragment::None,
68			label: None,
69			help: Some("The requested CDC version may have been garbage collected or never existed"
70				.to_string()),
71			notes: vec![],
72			cause: None,
73			operator_chain: None,
74		}
75	}
76
77	pub fn consumer_overtaken(consumer: impl Into<String>, cursor: u64, truncated_before: u64) -> Diagnostic {
78		Diagnostic {
79			code: "CDC_004".to_string(),
80			rql: None,
81			message: format!(
82				"CDC consumer '{}' was overtaken by retention: its cursor {} is below the truncation \
83				 floor {}, the changes in between are gone",
84				consumer.into(),
85				cursor,
86				truncated_before
87			),
88			column: None,
89			fragment: Fragment::None,
90			label: None,
91			help: Some("The consumer lagged past the CDC TTL window and must resync from a fresh \
92				    snapshot instead of resuming its cursor"
93				.to_string()),
94			notes: vec![],
95			cause: None,
96			operator_chain: None,
97		}
98	}
99
100	pub fn codec_error(msg: impl Into<String>) -> Diagnostic {
101		Diagnostic {
102			code: "CDC_003".to_string(),
103			rql: None,
104			message: format!("CDC codec error: {}", msg.into()),
105			column: None,
106			fragment: Fragment::None,
107			label: None,
108			help: Some("This may indicate data corruption or version mismatch".to_string()),
109			notes: vec![],
110			cause: None,
111			operator_chain: None,
112		}
113	}
114}
reifydb_cdc/error.rs

reifydb_cdc/
error.rs
