laminar_connectors/error/
mod.rs

1use thiserror::Error;
2
3/// Errors that can occur during connector operations.
4///
5/// Callers that need to distinguish "retry may work" from "propagate"
6/// should use [`ConnectorError::is_transient`] rather than matching
7/// variants directly — the variant set has changed in the past.
8#[derive(Debug, Error)]
9pub enum ConnectorError {
10    /// A connector factory was registered more than once in the same registry category.
11    #[error("{kind} connector factory '{name}' is already registered")]
12    FactoryAlreadyRegistered {
13        /// Registry category (`source`, `sink`, `table source`, or `lookup source`).
14        kind: &'static str,
15        /// Duplicate connector type name.
16        name: String,
17    },
18
19    /// Connector factory registration was attempted after construction completed.
20    #[error("connector registry is frozen; cannot register {kind} factory '{name}'")]
21    RegistryFrozen {
22        /// Registry category (`source`, `sink`, `table source`, or `lookup source`).
23        kind: &'static str,
24        /// Connector type name that was rejected.
25        name: String,
26    },
27
28    /// Failed to connect to the external system (network error, DNS
29    /// failure, TLS negotiation failure, auth rejection).
30    #[error("connection failed: {0}")]
31    ConnectionFailed(String),
32
33    /// Invalid, missing, or contradictory connector configuration.
34    #[error("configuration error: {0}")]
35    ConfigurationError(String),
36
37    /// A requested connector capability is intentionally unavailable.
38    #[error("feature unsupported: {0}")]
39    FeatureUnsupported(String),
40
41    /// Error reading data from a source.
42    #[error("read error: {0}")]
43    ReadError(String),
44
45    /// Error writing data to a sink.
46    #[error("write error: {0}")]
47    WriteError(String),
48
49    /// A dispatched operation failed without proving whether its external
50    /// side effects were applied.
51    ///
52    /// Recovery may replay the operation, but the connector generation that
53    /// observed this error must not process later work because its external
54    /// position is no longer known.
55    #[error("operation outcome unknown: {message}")]
56    OutcomeUnknown {
57        /// Failure detail.
58        message: String,
59        /// Whether a fresh connector generation may make progress after reconciliation.
60        retryable: bool,
61    },
62
63    /// Serialization or deserialization error.
64    #[error("serde error: {0}")]
65    Serde(#[from] SerdeError),
66
67    /// Transaction error (begin/commit/rollback).
68    ///
69    /// Kept separate from [`Self::WriteError`] because transactional
70    /// failures are classified as **non-transient** by default; a write
71    /// error is transient. Per-connector retry policy can override, but
72    /// the default must not loop forever on bad transactional state.
73    #[error("transaction error: {0}")]
74    TransactionError(String),
75
76    /// The connector is not in the expected state.
77    #[error("invalid state: expected {expected}, got {actual}")]
78    InvalidState {
79        /// The expected state.
80        expected: String,
81        /// The actual state.
82        actual: String,
83    },
84
85    /// Schema mismatch between expected and actual data.
86    #[error("schema mismatch: {0}")]
87    SchemaMismatch(String),
88
89    /// Operation timed out.
90    #[error("timeout after {0}ms")]
91    Timeout(u64),
92
93    /// The connector has been closed.
94    #[error("connector closed")]
95    Closed,
96
97    /// An internal error that doesn't fit other categories.
98    #[error("internal error: {0}")]
99    Internal(String),
100
101    /// An I/O error from the underlying system.
102    #[error("I/O error: {0}")]
103    Io(#[from] std::io::Error),
104}
105
106impl From<laminar_core::lookup::source::LookupError> for ConnectorError {
107    fn from(err: laminar_core::lookup::source::LookupError) -> Self {
108        use laminar_core::lookup::source::LookupError;
109        match err {
110            LookupError::Connection(m) => Self::ConnectionFailed(m),
111            LookupError::Query(m) => Self::ReadError(m),
112            LookupError::Timeout(d) =>
113            {
114                #[allow(clippy::cast_possible_truncation)]
115                Self::Timeout(d.as_millis() as u64)
116            }
117            LookupError::NotAvailable(m) => Self::InvalidState {
118                expected: "lookup source available".into(),
119                actual: m,
120            },
121            LookupError::Internal(m) => Self::Internal(m),
122        }
123    }
124}
125
126impl ConnectorError {
127    /// Construct an error for a dispatched operation whose external outcome is unknown.
128    #[must_use]
129    pub fn outcome_unknown(message: impl Into<String>, retryable: bool) -> Self {
130        Self::OutcomeUnknown {
131            message: message.into(),
132            retryable,
133        }
134    }
135
136    /// Construct a "missing required config" error. Thin helper around
137    /// [`Self::ConfigurationError`] so every "missing required config:
138    /// {key}" message is shaped the same way.
139    #[must_use]
140    pub fn missing_config(key: impl Into<String>) -> Self {
141        Self::ConfigurationError(format!("missing required config: {}", key.into()))
142    }
143
144    /// Returns `true` if this error is likely transient and recovery or a
145    /// retry may make progress (e.g., network timeout, throttled request).
146    /// An outcome-unknown error still requires connector retirement before
147    /// replay; inspect [`Self::is_outcome_unknown`] when that distinction
148    /// matters.
149    /// Returns `false` for configuration, schema, and state errors that
150    /// will not resolve without user intervention.
151    #[must_use]
152    pub fn is_transient(&self) -> bool {
153        match self {
154            Self::OutcomeUnknown { retryable, .. } => *retryable,
155            Self::ReadError(_)
156            | Self::WriteError(_)
157            | Self::Timeout(_)
158            | Self::Io(_)
159            | Self::ConnectionFailed(_) => true,
160
161            Self::ConfigurationError(_)
162            | Self::FeatureUnsupported(_)
163            | Self::FactoryAlreadyRegistered { .. }
164            | Self::RegistryFrozen { .. }
165            | Self::SchemaMismatch(_)
166            | Self::InvalidState { .. }
167            | Self::TransactionError(_)
168            | Self::Serde(_)
169            | Self::Closed
170            | Self::Internal(_) => false,
171        }
172    }
173
174    /// Returns `true` when an external side effect may have completed even
175    /// though the operation returned an error.
176    ///
177    /// This is stronger than [`Self::is_transient`]: callers may recover and
178    /// replay, but must retire the connector generation before doing so.
179    #[must_use]
180    pub fn is_outcome_unknown(&self) -> bool {
181        matches!(self, Self::OutcomeUnknown { .. })
182    }
183}
184
185/// Errors that occur during record serialization or deserialization.
186#[derive(Debug, Error)]
187pub enum SerdeError {
188    /// JSON parsing or encoding error.
189    #[error("JSON error: {0}")]
190    Json(String),
191
192    /// CSV parsing or encoding error.
193    #[error("CSV error: {0}")]
194    Csv(String),
195
196    /// The data format is not supported.
197    #[error("unsupported format: {0}")]
198    UnsupportedFormat(String),
199
200    /// A required field is missing from the input.
201    #[error("missing field: {0}")]
202    MissingField(String),
203
204    /// A field value could not be converted to the target Arrow type.
205    #[error("type conversion error: field '{field}', expected {expected}: {message}")]
206    TypeConversion {
207        /// The field name.
208        field: String,
209        /// The expected Arrow data type.
210        expected: String,
211        /// Details about the conversion failure.
212        message: String,
213    },
214
215    /// The input data is malformed.
216    #[error("malformed input: {0}")]
217    MalformedInput(String),
218
219    /// Schema ID not found in registry.
220    #[error("schema not found: schema ID {schema_id}")]
221    SchemaNotFound {
222        /// The schema ID that was not found.
223        schema_id: i32,
224    },
225
226    /// Confluent wire format magic byte mismatch.
227    #[error("invalid Confluent header: expected 0x{expected:02x}, got 0x{got:02x}")]
228    InvalidConfluentHeader {
229        /// Expected magic byte (0x00).
230        expected: u8,
231        /// Actual byte found.
232        got: u8,
233    },
234
235    /// Schema incompatible with existing version in the registry.
236    #[error("schema incompatible: subject '{subject}': {message}")]
237    SchemaIncompatible {
238        /// The Schema Registry subject name.
239        subject: String,
240        /// Incompatibility details.
241        message: String,
242    },
243
244    /// Avro decode failure for a specific column.
245    #[error("Avro decode error: column '{column}' (avro type '{avro_type}'): {message}")]
246    AvroDecodeError {
247        /// The column that failed to decode.
248        column: String,
249        /// The Avro type being decoded.
250        avro_type: String,
251        /// The decode failure details.
252        message: String,
253    },
254
255    /// Record count mismatch after serialization.
256    #[error("record count mismatch: expected {expected}, got {got}")]
257    RecordCountMismatch {
258        /// Expected number of records.
259        expected: usize,
260        /// Actual number of records produced.
261        got: usize,
262    },
263}
264
265impl From<serde_json::Error> for SerdeError {
266    fn from(e: serde_json::Error) -> Self {
267        SerdeError::Json(e.to_string())
268    }
269}
270
271#[cfg(test)]
272mod tests;
laminar_connectors/error/mod.rs

laminar_connectors/error/
mod.rs
