dataflow-rs 4.0.0

//! The error type and the wire shape of a recorded error.

use serde::ser::SerializeStruct;
use serde::{Deserialize, Serialize, Serializer};
use std::borrow::Cow;
use std::fmt;
use std::sync::Arc;

/// Why a run, a compile or a function failed.
pub struct Error {
    kind: ErrorKind,
    message: Cow<'static, str>,
    retryable: bool,
    source: Option<Box<dyn std::error::Error + Send + Sync>>,
}

/// The class of an [`Error`]. It decides the recorded code.
#[derive(Clone, Debug, PartialEq, Eq)]
#[non_exhaustive]
pub enum ErrorKind {
    /// A configuration is invalid.
    Config,
    /// An expression or definition did not compile.
    Compile,
    /// An expression failed to evaluate.
    Eval,
    /// The operation budget was exceeded.
    Budget,
    /// A deadline passed.
    TimedOut,
    /// The run was cancelled.
    Cancelled,
    /// A function reported its own code.
    Service {
        /// The code the function chose.
        code: Cow<'static, str>,
    },
}

/// The code of a task's own failure: a `Service` error with no code, and a
/// 5xx status's failure.
pub(crate) const TASK_ERROR: &str = "TASK_ERROR";
/// The record a 5xx status adds.
pub(crate) const TASK_STATUS_ERROR: &str = "TASK_STATUS_ERROR";
/// The record of a failed validation rule, and a built-in's refused input.
pub(crate) const VALIDATION_ERROR: &str = "VALIDATION_ERROR";
/// The record of a `map` mapping that failed, with its destination as the
/// record's `path`.
pub(crate) const MAPPING_ERROR: &str = "MAPPING_ERROR";
/// The record of a built-in's expression that failed to evaluate: a
/// `validation` rule, a `filter` condition.
pub(crate) const EVALUATION_ERROR: &str = "EVALUATION_ERROR";

impl Error {
    pub(crate) fn new(kind: ErrorKind, message: impl Into<Cow<'static, str>>) -> Self {
        Self {
            retryable: kind == ErrorKind::TimedOut,
            kind,
            message: message.into(),
            source: None,
        }
    }

    /// An error carrying a function's own code.
    pub fn service(
        code: impl Into<Cow<'static, str>>,
        message: impl Into<Cow<'static, str>>,
    ) -> Self {
        Self {
            kind: ErrorKind::Service { code: code.into() },
            message: message.into(),
            retryable: false,
            source: None,
        }
    }

    /// The class of the error.
    pub fn kind(&self) -> &ErrorKind {
        &self.kind
    }

    /// The human-readable message.
    pub fn message(&self) -> &str {
        &self.message
    }

    /// Overrides whether a retry may succeed.
    #[must_use]
    pub fn with_retryable(mut self, yes: bool) -> Self {
        self.retryable = yes;
        self
    }

    /// Attaches the error that caused this one.
    #[must_use]
    pub fn with_source(mut self, source: impl std::error::Error + Send + Sync + 'static) -> Self {
        self.source = Some(Box::new(source));
        self
    }

    /// May a retry succeed? A timeout says yes unless told otherwise; everything
    /// else says no.
    pub fn is_retryable(&self) -> bool {
        self.retryable
    }

    /// Prefixes the message with `context`, keeping kind, retryability and source.
    #[must_use]
    pub fn wrap(mut self, context: impl fmt::Display) -> Self {
        self.message = Cow::Owned(format!("{context}: {}", self.message));
        self
    }

    /// The recorded code for this error.
    pub fn code(&self) -> &str {
        match &self.kind {
            ErrorKind::Service { code } if !code.is_empty() => code,
            _ => self.code_static(),
        }
    }

    /// The static code for any non-service error, or for a service error with
    /// an empty code.
    pub(crate) fn code_static(&self) -> &'static str {
        match &self.kind {
            ErrorKind::Eval => "LOGIC_ERROR",
            ErrorKind::Budget => "BUDGET_EXCEEDED",
            ErrorKind::TimedOut => "TIMEOUT_ERROR",
            ErrorKind::Cancelled => "CANCELLED",
            ErrorKind::Config => "CONFIG_ERROR",
            ErrorKind::Compile => "COMPILATION_ERROR",
            ErrorKind::Service { .. } => TASK_ERROR,
        }
    }

    /// The text of this error's record: what v3's `Display` prints for the
    /// variant `kind` corresponds to, which the recorder wraps as
    /// "Task {id} error: {text}". A kind v3 has no variant for, and a service
    /// error (v3's display is the bare message), give the message alone.
    pub(crate) fn record_text(&self) -> Cow<'_, str> {
        let prefix = match &self.kind {
            ErrorKind::Eval => "Logic evaluation error: ",
            ErrorKind::TimedOut => "Timeout error: ",
            ErrorKind::Budget
            | ErrorKind::Config
            | ErrorKind::Compile
            | ErrorKind::Cancelled
            | ErrorKind::Service { .. } => return Cow::Borrowed(&self.message),
        };
        Cow::Owned(format!("{prefix}{}", self.message))
    }
}

impl fmt::Display for Error {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.write_str(&self.message)
    }
}

impl fmt::Debug for Error {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("Error")
            .field("kind", &self.kind)
            .field("message", &self.message)
            .field("retryable", &self.is_retryable())
            .field("source", &self.source)
            .finish_non_exhaustive()
    }
}

impl std::error::Error for Error {
    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
        self.source.as_deref().map(|s| s as _)
    }
}

/// The result of a fallible v4 operation.
pub type Result<T> = std::result::Result<T, Error>;

/// One error record, written as v3 wrote one: the same keys, in the same
/// order, each optional one absent when unset. The engine stamps every
/// record a task adds with the workflow's and the task's ids and the run's
/// timestamp, keeping any the record already carries; a workflow's own
/// failure carries no task id.
#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
#[cfg_attr(feature = "ts", derive(ts_rs::TS))]
#[non_exhaustive]
pub struct ErrorRecord {
    /// The error code.
    pub code: String,
    /// The human-readable message.
    pub message: String,
    /// Where the error happened; `null` when not known, and never absent.
    pub path: Option<String>,
    /// The workflow it happened in, shared with the program's own id and
    /// the audit entry's, so a run allocates it once.
    #[serde(skip_serializing_if = "Option::is_none")]
    #[cfg_attr(feature = "ts", ts(optional))]
    pub workflow_id: Option<Arc<str>>,
    /// The task it happened in, shared as `workflow_id` is.
    #[serde(skip_serializing_if = "Option::is_none")]
    #[cfg_attr(feature = "ts", ts(optional))]
    pub task_id: Option<Arc<str>>,
    /// When it happened, shared with every record of the run.
    #[serde(skip_serializing_if = "Option::is_none")]
    #[cfg_attr(feature = "ts", ts(optional))]
    pub timestamp: Option<Arc<str>>,
    /// Operator-only detail; never to be shown to an untrusted caller.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    #[cfg_attr(feature = "ts", ts(optional))]
    pub detail: Option<String>,
    /// The fan-out element the error came from.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    #[cfg_attr(feature = "ts", ts(optional))]
    pub element_index: Option<usize>,
}

impl ErrorRecord {
    /// A record with a code and a message and nothing else.
    pub fn new(code: impl Into<String>, message: impl Into<String>) -> Self {
        Self {
            code: code.into(),
            message: message.into(),
            path: None,
            workflow_id: None,
            task_id: None,
            timestamp: None,
            detail: None,
            element_index: None,
        }
    }

    /// Fills what the record leaves unset: the workflow's id, the task's id
    /// (`None` for a workflow's own failure), the run's timestamp `now`,
    /// and in a fan-out the element's index. Every record the engine keeps
    /// is stamped here, so a consumer can group records by task. The ids and
    /// the timestamp are the program's and the run's own `Arc<str>`, shared
    /// rather than copied.
    pub(crate) fn stamp(
        &mut self,
        workflow_id: &Arc<str>,
        task_id: Option<&Arc<str>>,
        now: &Arc<str>,
        element_index: Option<usize>,
    ) {
        if self.workflow_id.is_none() {
            self.workflow_id = Some(Arc::clone(workflow_id));
        }
        if self.task_id.is_none() {
            self.task_id = task_id.map(Arc::clone);
        }
        if self.timestamp.is_none() {
            self.timestamp = Some(Arc::clone(now));
        }
        if self.element_index.is_none() {
            self.element_index = element_index;
        }
    }
}

impl From<&Error> for ErrorRecord {
    fn from(error: &Error) -> Self {
        Self::new(error.code(), error.message())
    }
}

/// Writes the error as `{"code", "message", "retryable"}`: [`Error::code`],
/// [`Error::message`] and [`Error::is_retryable`], in that order.
impl Serialize for Error {
    fn serialize<S: Serializer>(&self, s: S) -> std::result::Result<S::Ok, S::Error> {
        let mut o = s.serialize_struct("Error", 3)?;
        o.serialize_field("code", self.code())?;
        o.serialize_field("message", self.message())?;
        o.serialize_field("retryable", &self.is_retryable())?;
        o.end()
    }
}

// `DataflowError`, not `Error`: a TypeScript `Error` would shadow the
// global one wherever it is imported.
#[cfg(feature = "ts")]
crate::base::ts::ts_decl!(
    Error,
    "DataflowError",
    [
        " An error as JSON: its code, its message, and whether a retry may",
        " succeed.",
    ],
    |_cfg| {
        use crate::base::ts::{Prop, object};
        object(&[
            Prop::new("code", "string", &[" The recorded code."]),
            Prop::new("message", "string", &[" The human-readable message."]),
            Prop::new("retryable", "boolean", &[" May a retry succeed?"]),
        ])
    }
);

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn record_text_prefixes_two_kinds_and_is_the_message_otherwise() {
        // The record texts are 3.x's, byte for byte.
        let m = "the message";
        for (kind, expected) in [
            (ErrorKind::Eval, "Logic evaluation error: the message"),
            (ErrorKind::TimedOut, "Timeout error: the message"),
            (ErrorKind::Budget, m),
            (ErrorKind::Service { code: "X".into() }, m),
            (ErrorKind::Config, m),
            (ErrorKind::Compile, m),
            (ErrorKind::Cancelled, m),
        ] {
            assert_eq!(
                Error::new(kind.clone(), m).record_text(),
                expected,
                "{kind:?}"
            );
        }
    }

    fn kinds() -> Vec<(ErrorKind, &'static str)> {
        vec![
            (ErrorKind::Eval, "LOGIC_ERROR"),
            (ErrorKind::Budget, "BUDGET_EXCEEDED"),
            (ErrorKind::TimedOut, "TIMEOUT_ERROR"),
            (ErrorKind::Cancelled, "CANCELLED"),
            (ErrorKind::Config, "CONFIG_ERROR"),
            (ErrorKind::Compile, "COMPILATION_ERROR"),
            (
                ErrorKind::Service {
                    code: "X_FAIL".into(),
                },
                "X_FAIL",
            ),
            (ErrorKind::Service { code: "".into() }, "TASK_ERROR"),
        ]
    }

    #[test]
    fn code_gives_each_row_of_the_table() {
        for (kind, code) in kinds() {
            assert_eq!(Error::new(kind, "m").code(), code);
        }
    }

    #[test]
    fn only_a_timeout_is_retryable_by_default() {
        for (kind, _) in kinds() {
            let timed_out = kind == ErrorKind::TimedOut;
            assert_eq!(Error::new(kind, "m").is_retryable(), timed_out);
        }
        assert!(Error::service("A", "m").with_retryable(true).is_retryable());
        assert!(
            !Error::new(ErrorKind::TimedOut, "m")
                .with_retryable(false)
                .is_retryable()
        );
    }

    #[test]
    fn wrap_keeps_kind_retryable_and_source() {
        let source = std::io::Error::other("disk");
        let error = Error::new(ErrorKind::Eval, "bad")
            .with_retryable(true)
            .with_source(source)
            .wrap("task t");
        assert_eq!(error.to_string(), "task t: bad");
        assert_eq!(error.kind(), &ErrorKind::Eval);
        assert!(error.is_retryable());
        let source = std::error::Error::source(&error).expect("source kept");
        assert_eq!(source.to_string(), "disk");
    }

    #[test]
    fn debug_shows_kind_message_and_retryable() {
        let text = format!("{:?}", Error::new(ErrorKind::Budget, "over"));
        assert!(text.contains("Budget") && text.contains("over") && text.contains("retryable"));
    }

    #[test]
    fn a_stored_record_with_3x_retry_keys_reads_and_writes_without_them() {
        let stored =
            r#"{"code":"X","message":"m","path":null,"retry_attempted":true,"retry_count":2}"#;
        let record: ErrorRecord = serde_json::from_str(stored).unwrap();
        assert_eq!(
            serde_json::to_string(&record).unwrap(),
            r#"{"code":"X","message":"m","path":null}"#
        );
    }

    #[test]
    fn stamp_fills_only_what_the_record_leaves_unset() {
        let mut record = ErrorRecord::new("X", "m");
        record.task_id = Some(Arc::from("own"));
        record.stamp(
            &Arc::from("w"),
            Some(&Arc::from("t")),
            &Arc::from("now"),
            Some(3),
        );
        assert_eq!(
            serde_json::to_string(&record).unwrap(),
            r#"{"code":"X","message":"m","path":null,"workflow_id":"w","task_id":"own","timestamp":"now","element_index":3}"#
        );
        let mut workflow_own = ErrorRecord::new("X", "m");
        workflow_own.stamp(&Arc::from("w"), None, &Arc::from("now"), None);
        assert_eq!(workflow_own.task_id, None);
    }

    #[test]
    fn a_record_from_an_error_carries_only_code_and_message() {
        let record = ErrorRecord::from(&Error::service("X", "boom"));
        assert_eq!(
            serde_json::to_string(&record).unwrap(),
            r#"{"code":"X","message":"boom","path":null}"#
        );
    }
}