reifydb-cdc 0.4.5

Change Data Capture module for ReifyDB
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100

// SPDX-License-Identifier: Apache-2.0
// Copyright (c) 2025 ReifyDB

//! CDC error types and diagnostics.

use std::{error::Error as StdError, fmt, fmt::Display};

use reifydb_core::common::CommitVersion;
use reifydb_type::{error, error::Error};

/// Error type for CDC operations.
#[derive(Debug, Clone)]
pub enum CdcError {
	/// The operation failed due to an internal error.
	Internal(String),
	/// The CDC entry was not found.
	NotFound(CommitVersion),
	/// Encoding or decoding failed.
	Codec(String),
}

impl Display for CdcError {
	fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
		match self {
			CdcError::Internal(msg) => write!(f, "CDC storage internal error: {}", msg),
			CdcError::NotFound(version) => write!(f, "CDC entry not found: {:?}", version),
			CdcError::Codec(msg) => write!(f, "CDC codec error: {}", msg),
		}
	}
}

impl StdError for CdcError {}

impl From<CdcError> for Error {
	fn from(err: CdcError) -> Self {
		error!(match err {
			CdcError::Internal(msg) => diagnostic::storage_error(msg),
			CdcError::NotFound(version) => diagnostic::not_found(version.0),
			CdcError::Codec(msg) => diagnostic::codec_error(msg),
		})
	}
}

/// Result type for CDC operations.
pub type CdcResult<T> = Result<T, CdcError>;

/// CDC-specific diagnostics.
pub mod diagnostic {

	use reifydb_type::{error::Diagnostic, fragment::Fragment};

	/// CDC storage operation failed
	pub fn storage_error(msg: impl Into<String>) -> Diagnostic {
		Diagnostic {
			code: "CDC_001".to_string(),
			statement: None,
			message: format!("CDC storage error: {}", msg.into()),
			column: None,
			fragment: Fragment::None,
			label: None,
			help: Some("Check CDC storage configuration and availability".to_string()),
			notes: vec![],
			cause: None,
			operator_chain: None,
		}
	}

	/// CDC entry not found at the specified version
	pub fn not_found(version: u64) -> Diagnostic {
		Diagnostic {
			code: "CDC_002".to_string(),
			statement: None,
			message: format!("CDC entry not found for version {}", version),
			column: None,
			fragment: Fragment::None,
			label: None,
			help: Some("The requested CDC version may have been garbage collected or never existed"
				.to_string()),
			notes: vec![],
			cause: None,
			operator_chain: None,
		}
	}

	/// CDC encoding/decoding failed
	pub fn codec_error(msg: impl Into<String>) -> Diagnostic {
		Diagnostic {
			code: "CDC_003".to_string(),
			statement: None,
			message: format!("CDC codec error: {}", msg.into()),
			column: None,
			fragment: Fragment::None,
			label: None,
			help: Some("This may indicate data corruption or version mismatch".to_string()),
			notes: vec![],
			cause: None,
			operator_chain: None,
		}
	}
}