laminar_connectors/
error.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    /// Error reading data from a source.
38    #[error("read error: {0}")]
39    ReadError(String),
40
41    /// Error writing data to a sink.
42    #[error("write error: {0}")]
43    WriteError(String),
44
45    /// A dispatched operation failed without proving whether its external
46    /// side effects were applied.
47    ///
48    /// Recovery may replay the operation, but the connector generation that
49    /// observed this error must not process later work because its external
50    /// position is no longer known.
51    #[error("operation outcome unknown: {message}")]
52    OutcomeUnknown {
53        /// Failure detail.
54        message: String,
55        /// Whether a fresh connector generation may make progress after reconciliation.
56        retryable: bool,
57    },
58
59    /// Serialization or deserialization error.
60    #[error("serde error: {0}")]
61    Serde(#[from] SerdeError),
62
63    /// Transaction error (begin/commit/rollback).
64    ///
65    /// Kept separate from [`Self::WriteError`] because transactional
66    /// failures are classified as **non-transient** by default; a write
67    /// error is transient. Per-connector retry policy can override, but
68    /// the default must not loop forever on bad transactional state.
69    #[error("transaction error: {0}")]
70    TransactionError(String),
71
72    /// The connector is not in the expected state.
73    #[error("invalid state: expected {expected}, got {actual}")]
74    InvalidState {
75        /// The expected state.
76        expected: String,
77        /// The actual state.
78        actual: String,
79    },
80
81    /// Schema mismatch between expected and actual data.
82    #[error("schema mismatch: {0}")]
83    SchemaMismatch(String),
84
85    /// Operation timed out.
86    #[error("timeout after {0}ms")]
87    Timeout(u64),
88
89    /// The connector has been closed.
90    #[error("connector closed")]
91    Closed,
92
93    /// An internal error that doesn't fit other categories.
94    #[error("internal error: {0}")]
95    Internal(String),
96
97    /// An I/O error from the underlying system.
98    #[error("I/O error: {0}")]
99    Io(#[from] std::io::Error),
100}
101
102impl From<laminar_core::lookup::source::LookupError> for ConnectorError {
103    fn from(err: laminar_core::lookup::source::LookupError) -> Self {
104        use laminar_core::lookup::source::LookupError;
105        match err {
106            LookupError::Connection(m) => Self::ConnectionFailed(m),
107            LookupError::Query(m) => Self::ReadError(m),
108            LookupError::Timeout(d) =>
109            {
110                #[allow(clippy::cast_possible_truncation)]
111                Self::Timeout(d.as_millis() as u64)
112            }
113            LookupError::NotAvailable(m) => Self::InvalidState {
114                expected: "lookup source available".into(),
115                actual: m,
116            },
117            LookupError::Internal(m) => Self::Internal(m),
118        }
119    }
120}
121
122impl ConnectorError {
123    /// Construct an error for a dispatched operation whose external outcome is unknown.
124    #[must_use]
125    pub fn outcome_unknown(message: impl Into<String>, retryable: bool) -> Self {
126        Self::OutcomeUnknown {
127            message: message.into(),
128            retryable,
129        }
130    }
131
132    /// Construct a "missing required config" error. Thin helper around
133    /// [`Self::ConfigurationError`] so every "missing required config:
134    /// {key}" message is shaped the same way.
135    #[must_use]
136    pub fn missing_config(key: impl Into<String>) -> Self {
137        Self::ConfigurationError(format!("missing required config: {}", key.into()))
138    }
139
140    /// Returns `true` if this error is likely transient and recovery or a
141    /// retry may make progress (e.g., network timeout, throttled request).
142    /// An outcome-unknown error still requires connector retirement before
143    /// replay; inspect [`Self::is_outcome_unknown`] when that distinction
144    /// matters.
145    /// Returns `false` for configuration, schema, and state errors that
146    /// will not resolve without user intervention.
147    #[must_use]
148    pub fn is_transient(&self) -> bool {
149        match self {
150            Self::OutcomeUnknown { retryable, .. } => *retryable,
151            Self::ReadError(_)
152            | Self::WriteError(_)
153            | Self::Timeout(_)
154            | Self::Io(_)
155            | Self::ConnectionFailed(_) => true,
156
157            Self::ConfigurationError(_)
158            | Self::FactoryAlreadyRegistered { .. }
159            | Self::RegistryFrozen { .. }
160            | Self::SchemaMismatch(_)
161            | Self::InvalidState { .. }
162            | Self::TransactionError(_)
163            | Self::Serde(_)
164            | Self::Closed
165            | Self::Internal(_) => false,
166        }
167    }
168
169    /// Returns `true` when an external side effect may have completed even
170    /// though the operation returned an error.
171    ///
172    /// This is stronger than [`Self::is_transient`]: callers may recover and
173    /// replay, but must retire the connector generation before doing so.
174    #[must_use]
175    pub fn is_outcome_unknown(&self) -> bool {
176        matches!(self, Self::OutcomeUnknown { .. })
177    }
178}
179
180/// Errors that occur during record serialization or deserialization.
181#[derive(Debug, Error)]
182pub enum SerdeError {
183    /// JSON parsing or encoding error.
184    #[error("JSON error: {0}")]
185    Json(String),
186
187    /// CSV parsing or encoding error.
188    #[error("CSV error: {0}")]
189    Csv(String),
190
191    /// The data format is not supported.
192    #[error("unsupported format: {0}")]
193    UnsupportedFormat(String),
194
195    /// A required field is missing from the input.
196    #[error("missing field: {0}")]
197    MissingField(String),
198
199    /// A field value could not be converted to the target Arrow type.
200    #[error("type conversion error: field '{field}', expected {expected}: {message}")]
201    TypeConversion {
202        /// The field name.
203        field: String,
204        /// The expected Arrow data type.
205        expected: String,
206        /// Details about the conversion failure.
207        message: String,
208    },
209
210    /// The input data is malformed.
211    #[error("malformed input: {0}")]
212    MalformedInput(String),
213
214    /// Schema ID not found in registry.
215    #[error("schema not found: schema ID {schema_id}")]
216    SchemaNotFound {
217        /// The schema ID that was not found.
218        schema_id: i32,
219    },
220
221    /// Confluent wire format magic byte mismatch.
222    #[error("invalid Confluent header: expected 0x{expected:02x}, got 0x{got:02x}")]
223    InvalidConfluentHeader {
224        /// Expected magic byte (0x00).
225        expected: u8,
226        /// Actual byte found.
227        got: u8,
228    },
229
230    /// Schema incompatible with existing version in the registry.
231    #[error("schema incompatible: subject '{subject}': {message}")]
232    SchemaIncompatible {
233        /// The Schema Registry subject name.
234        subject: String,
235        /// Incompatibility details.
236        message: String,
237    },
238
239    /// Avro decode failure for a specific column.
240    #[error("Avro decode error: column '{column}' (avro type '{avro_type}'): {message}")]
241    AvroDecodeError {
242        /// The column that failed to decode.
243        column: String,
244        /// The Avro type being decoded.
245        avro_type: String,
246        /// The decode failure details.
247        message: String,
248    },
249
250    /// Record count mismatch after serialization.
251    #[error("record count mismatch: expected {expected}, got {got}")]
252    RecordCountMismatch {
253        /// Expected number of records.
254        expected: usize,
255        /// Actual number of records produced.
256        got: usize,
257    },
258}
259
260impl From<serde_json::Error> for SerdeError {
261    fn from(e: serde_json::Error) -> Self {
262        SerdeError::Json(e.to_string())
263    }
264}
265
266#[cfg(test)]
267mod tests {
268    use super::*;
269
270    #[test]
271    fn test_connector_error_display() {
272        let err = ConnectorError::ConnectionFailed("host unreachable".into());
273        assert_eq!(err.to_string(), "connection failed: host unreachable");
274    }
275
276    #[test]
277    fn test_serde_error_from_json() {
278        let json_err: Result<serde_json::Value, _> = serde_json::from_str("{bad json");
279        let serde_err: SerdeError = json_err.unwrap_err().into();
280        assert!(matches!(serde_err, SerdeError::Json(_)));
281    }
282
283    #[test]
284    fn test_serde_error_into_connector_error() {
285        let serde_err = SerdeError::MissingField("timestamp".into());
286        let conn_err: ConnectorError = serde_err.into();
287        assert!(matches!(conn_err, ConnectorError::Serde(_)));
288        assert!(conn_err.to_string().contains("timestamp"));
289    }
290
291    #[test]
292    fn test_invalid_state_error() {
293        let err = ConnectorError::InvalidState {
294            expected: "Running".into(),
295            actual: "Closed".into(),
296        };
297        assert!(err.to_string().contains("Running"));
298        assert!(err.to_string().contains("Closed"));
299    }
300
301    #[test]
302    fn outcome_unknown_is_recoverable_but_requires_retirement() {
303        let err = ConnectorError::outcome_unknown("publish acknowledgement timed out", true);
304        assert!(err.is_transient());
305        assert!(err.is_outcome_unknown());
306        assert_eq!(
307            err.to_string(),
308            "operation outcome unknown: publish acknowledgement timed out"
309        );
310
311        assert!(!ConnectorError::Timeout(10).is_outcome_unknown());
312        assert!(
313            !ConnectorError::WriteError("rejected before dispatch".into()).is_outcome_unknown()
314        );
315
316        let terminal =
317            ConnectorError::outcome_unknown("permanent rejection after partial work", false);
318        assert!(terminal.is_outcome_unknown());
319        assert!(!terminal.is_transient());
320    }
321
322    #[test]
323    fn test_schema_not_found_error() {
324        let err = SerdeError::SchemaNotFound { schema_id: 42 };
325        assert!(err.to_string().contains("42"));
326        assert!(err.to_string().contains("schema not found"));
327    }
328
329    #[test]
330    fn test_invalid_confluent_header_error() {
331        let err = SerdeError::InvalidConfluentHeader {
332            expected: 0x00,
333            got: 0xFF,
334        };
335        let msg = err.to_string();
336        assert!(msg.contains("0x00"));
337        assert!(msg.contains("0xff"));
338    }
339
340    #[test]
341    fn test_schema_incompatible_error() {
342        let err = SerdeError::SchemaIncompatible {
343            subject: "orders-value".into(),
344            message: "READER_FIELD_MISSING_DEFAULT_VALUE".into(),
345        };
346        let msg = err.to_string();
347        assert!(msg.contains("orders-value"));
348        assert!(msg.contains("READER_FIELD_MISSING_DEFAULT_VALUE"));
349    }
350
351    #[test]
352    fn test_avro_decode_error() {
353        let err = SerdeError::AvroDecodeError {
354            column: "price".into(),
355            avro_type: "double".into(),
356            message: "unexpected null".into(),
357        };
358        let msg = err.to_string();
359        assert!(msg.contains("price"));
360        assert!(msg.contains("double"));
361        assert!(msg.contains("unexpected null"));
362    }
363
364    #[test]
365    fn test_record_count_mismatch_error() {
366        let err = SerdeError::RecordCountMismatch {
367            expected: 5,
368            got: 3,
369        };
370        let msg = err.to_string();
371        assert!(msg.contains('5'));
372        assert!(msg.contains('3'));
373    }
374}
laminar_connectors/error.rs

laminar_connectors/
error.rs
