use dataflow_rs::DataflowError;
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum ErrorClass {
CallerInput,
Limit,
Connector,
Backend,
Timeout,
}
impl ErrorClass {
pub fn is_retryable(self) -> bool {
matches!(self, ErrorClass::Connector | ErrorClass::Timeout)
}
}
#[derive(Debug, Clone)]
pub struct HandlerError {
pub class: ErrorClass,
pub msg: String,
pub detail: Option<String>,
original: Option<Box<DataflowError>>,
}
impl HandlerError {
pub fn new(class: ErrorClass, msg: impl std::fmt::Display) -> Self {
Self {
class,
msg: msg.to_string(),
detail: None,
original: None,
}
}
pub fn with_detail(mut self, detail: impl std::fmt::Display) -> Self {
self.detail = Some(detail.to_string());
self
}
pub fn prefixed(mut self, handler: &str) -> Self {
self.msg = format!("{handler}: {}", self.msg);
self.original = None;
self
}
}
impl std::fmt::Display for HandlerError {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.write_str(&self.msg)
}
}
impl From<HandlerError> for DataflowError {
fn from(e: HandlerError) -> Self {
if let Some(original) = e.original {
return *original;
}
match e.class {
ErrorClass::CallerInput | ErrorClass::Limit => DataflowError::Validation(e.msg),
ErrorClass::Connector => DataflowError::Io(e.msg),
ErrorClass::Backend => DataflowError::function_execution(e.msg, None),
ErrorClass::Timeout => DataflowError::Timeout(e.msg),
}
}
}
impl From<DataflowError> for HandlerError {
fn from(e: DataflowError) -> Self {
let detail = match &e {
DataflowError::Service { detail, .. } => detail.clone(),
_ => None,
};
let original = e.clone();
let (class, msg) = match e {
DataflowError::Validation(m) => (ErrorClass::CallerInput, m),
DataflowError::Timeout(m) => (ErrorClass::Timeout, m),
DataflowError::Io(m) => (ErrorClass::Connector, m),
DataflowError::FunctionExecution { context, .. } => (ErrorClass::Backend, context),
DataflowError::Service {
message, retryable, ..
} => (
if retryable {
ErrorClass::Connector
} else {
ErrorClass::Backend
},
message,
),
DataflowError::Http { status, message } => (
if (400..500).contains(&status) {
ErrorClass::CallerInput
} else {
ErrorClass::Backend
},
message,
),
DataflowError::Workflow(m)
| DataflowError::Task(m)
| DataflowError::FunctionNotFound(m)
| DataflowError::Deserialization(m)
| DataflowError::LogicEvaluation(m)
| DataflowError::Unknown(m) => (ErrorClass::Backend, m),
other => (ErrorClass::Backend, other.to_string()),
};
Self {
class,
msg,
detail,
original: Some(Box::new(original)),
}
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn each_class_maps_to_the_variant_its_constructor_used_to_pick() {
let cases = [
(ErrorClass::CallerInput, false),
(ErrorClass::Limit, false),
(ErrorClass::Connector, true),
(ErrorClass::Backend, false),
(ErrorClass::Timeout, true),
];
for (class, retryable) in cases {
assert_eq!(
class.is_retryable(),
retryable,
"{class:?} changed its retry policy"
);
}
}
#[test]
fn the_variant_mapping_is_the_one_the_three_constructors_made() {
let of = |class| DataflowError::from(HandlerError::new(class, "boom"));
assert!(matches!(
of(ErrorClass::CallerInput),
DataflowError::Validation(_)
));
assert!(matches!(
of(ErrorClass::Limit),
DataflowError::Validation(_)
));
assert!(matches!(of(ErrorClass::Connector), DataflowError::Io(_)));
assert!(matches!(
of(ErrorClass::Backend),
DataflowError::FunctionExecution { .. }
));
assert!(matches!(of(ErrorClass::Timeout), DataflowError::Timeout(_)));
}
#[test]
fn a_caller_fixable_failure_keeps_its_message() {
let err = DataflowError::from(HandlerError::new(
ErrorClass::Limit,
"add a LIMIT to the query or raise the cap",
));
let DataflowError::Validation(msg) = err else {
unreachable!("Limit maps to Validation")
};
assert_eq!(msg, "add a LIMIT to the query or raise the cap");
}
#[test]
fn prefixing_is_separable_from_the_message() {
let bare = HandlerError::new(ErrorClass::CallerInput, "'from' is not a valid address");
assert_eq!(bare.msg, "'from' is not a valid address");
assert_eq!(
bare.prefixed("send_email").msg,
"send_email: 'from' is not a valid address"
);
}
}
#[cfg(test)]
mod round_trip_tests {
use super::*;
#[test]
fn a_service_error_survives_the_round_trip() {
let original = crate::errors::connector_detail_error("operation 'read' is disabled");
let back: DataflowError = HandlerError::from(original).into();
match back {
DataflowError::Service { kind, detail, .. } => {
assert_eq!(kind, crate::errors::kind::CONNECTOR_DETAIL);
assert_eq!(detail.as_deref(), Some("operation 'read' is disabled"));
}
other => unreachable!("a Service error must return as one, got {other:?}"),
}
}
#[test]
fn prefixing_gives_up_the_round_trip() {
let original = crate::errors::connector_detail_error("nope");
let back: DataflowError = HandlerError::from(original).prefixed("crypto").into();
assert!(
matches!(back, DataflowError::FunctionExecution { .. }),
"a rewritten Service error is rebuilt from its class, got {back:?}"
);
}
}