dbos 0.6.0

//! The error channel for the execution engine.
//!
//! [`Error`] is generic over what the *application* failed with, and that one decision shapes the
//! rest. The engine wraps the application rather than the application making room for the engine:
//! a workflow returning `dbos::Result<Receipt, CheckoutError>` fails with either
//! [`Error::Application`] carrying a `CheckoutError`, or one of the engine's own variants. The
//! application's error type stays DBOS-agnostic — two derives, no variant of ours inside it, no
//! trait it has to know about.
//!
//! **The generic parameter buys back a blanket conversion**, which is what makes that work.
//! `impl<E> From<E> for Error<E>` is accepted where `impl<E: std::error::Error> From<E> for Error`
//! is not: the overlap with `core`'s reflexive `impl<T> From<T> for T` would need `E = Error<E>`,
//! an infinite type the occurs check rules out. So `?` lifts an application's own error into ours
//! with no impl written by anyone. `anyhow` gives up `std::error::Error` to get the same thing;
//! here it costs a type parameter instead.
//!
//! The price is paid inside this crate rather than by its users: the blanket impl is why
//! [`Error::SystemDatabase`] cannot also be a `#[from]` — that pair *does* overlap, at
//! `Error<sysdb::Error>` — so engine code names the conversion: `.map_err(Error::from_execution)?`
//! for a call a workflow's own execution makes, so a cancellation reaches it as
//! [`Error::WorkflowCancelled`], and `.map_err(Error::SystemDatabase)?` for one that can't be
//! cancelled.

use crate::registry::QualifiedWorkflowName;
use crate::serialization::{PayloadEncoding, decode_error, encode_error};
use std::borrow::Cow;

/// The error type of a workflow or step with no failure of its own.
///
/// Named for what it makes the channel: `Error<EngineOnly>` carries the engine's own errors and
/// nothing else. Uninhabited, so the compiler knows [`Error::Application`] cannot be constructed,
/// which is what lets an engine error be carried into any workflow's channel by a total conversion
/// rather than by a panic waiting for an input that cannot arrive.
///
/// This is the default parameter, so bare `dbos::Error` means "a failure of DBOS's", and
/// `dbos::Result<T>` is the result of an engine operation. A `From<MyError>` bound failing against
/// `Error<EngineOnly>` is the compiler saying the workflow declared no application error type —
/// the fix is to return `dbos::Result<T, MyError>`.
#[derive(Debug, serde::Serialize, serde::Deserialize)]
pub enum EngineOnly {}

impl std::fmt::Display for EngineOnly {
    fn fmt(&self, _: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match *self {}
    }
}

impl std::error::Error for EngineOnly {}

/// What an application may fail with, inside [`Error::Application`].
///
/// Notably **not** `From<Error>`: an application error type knows nothing about this crate. The
/// bounds are only what the engine needs of it:
///
/// - `Serialize` + `DeserializeOwned` — a recorded failure has to survive a column, and decoding
///   needs a concrete type. This is the whole reason the error is a type parameter rather than a
///   trait object.
/// - `std::error::Error` — so a failure reads as a failure: `Display` for a log, `source` for a
///   caller walking the chain.
///
/// Implemented for anything meeting them, so an application writes no impl:
///
/// ```
/// #[derive(Debug, thiserror::Error, serde::Serialize, serde::Deserialize)]
/// enum CheckoutError {
///     #[error("the card was declined")]
///     CardDeclined,
/// }
///
/// async fn checkout(cart: u32) -> dbos::Result<u32, CheckoutError> {
///     Err(CheckoutError::CardDeclined)?
/// }
/// ```
pub trait DurableError:
    serde::Serialize + serde::de::DeserializeOwned + std::error::Error + 'static
{
}

impl<E> DurableError for E where
    E: serde::Serialize + serde::de::DeserializeOwned + std::error::Error + 'static
{
}

/// The result of a durable function, or of an engine operation.
///
/// `Result<T>` is the engine's own — it cannot carry an application failure, because
/// [`EngineOnly`] has no values. `Result<T, CheckoutError>` is a workflow's, and note the second
/// parameter is the *application's* error rather than the whole error type: it expands to
/// `std::result::Result<T, Error<CheckoutError>>`.
pub type Result<T, E = EngineOnly> = std::result::Result<T, Error<E>>;

/// Everything a durable function can fail with.
///
/// `#[non_exhaustive]` because more variants arrive with the features that raise them; matching
/// callers need a wildcard arm from the start rather than a breaking change later.
///
/// **Serializable, and that is load-bearing.** A failed workflow or step records the error it
/// failed with, and a replay has to give back *that error* rather than a description of it — the
/// same fidelity a successful result gets. Every payload is data this crate or the application
/// owns, so encoding costs a derive; the only fields that cannot survive a round trip are the
/// `serde_json::Error` sources, marked `#[serde(skip)]` and explained where they are declared.
#[derive(Debug, thiserror::Error, serde::Serialize, serde::Deserialize)]
#[non_exhaustive]
pub enum Error<E = EngineOnly> {
    /// A failure reported by the workflow or step body itself.
    ///
    /// The application's own error, held as itself rather than reduced to a description of one.
    /// The run that fails and the replay that reads the row back both produce this variant with an
    /// equal payload, which is the property the whole generic parameter exists for.
    #[error(transparent)]
    Application(E),

    /// An operation needing a running executor was called before [`launch`](crate::DBOS::launch).
    ///
    /// Names the operation: the useful half of this error is which call was too early, and a bare
    /// "not launched" makes the developer find it.
    #[error("cannot {operation} before DBOS is launched")]
    NotLaunched {
        /// The operation that was called too early.
        operation: Cow<'static, str>,
    },

    /// An operation that must happen before launch was called after it.
    ///
    /// Registration is the case that matters: the executor takes a snapshot of the registry at
    /// launch, so a workflow registered afterwards would be invisible to recovery and to dequeue.
    #[error("cannot {operation} after DBOS is launched")]
    AlreadyLaunched {
        /// The operation that was called too late.
        operation: Cow<'static, str>,
    },

    /// An operation that only makes sense inside a workflow was called outside one.
    ///
    /// [`set_event`](crate::set_event) is the archetype: it checkpoints its write under a step id,
    /// and outside a workflow there is no step-id sequence to record against — unlike a step,
    /// whose body can simply run plainly.
    #[error("{operation} must be called from within a workflow")]
    NotInWorkflow {
        /// The operation that needed a workflow around it.
        operation: Cow<'static, str>,
    },

    /// An operation that allocates a step id was called from inside a step.
    ///
    /// A step is a leaf: its checkpoint stands for everything the body did, so an id-allocating
    /// operation inside one would shift every later step onto the wrong replay slot. The same
    /// rule that makes a nested step a plain call makes this an error — there is no plain
    /// version of a durable write to degrade to.
    #[error("{operation} cannot be called from within a step")]
    InsideStep {
        /// The operation that was called inside a step.
        operation: Cow<'static, str>,
    },

    /// An operation that is not recorded as a step was called from inside a workflow.
    ///
    /// [`apply_schedules`](crate::DBOS::apply_schedules),
    /// [`backfill_schedule`](crate::DBOS::backfill_schedule) and
    /// [`trigger_schedule`](crate::DBOS::trigger_schedule) each write a batch with no checkpoint,
    /// so a replay of the workflow would write it again. Refusing the call is what keeps a
    /// workflow's replay from repeating those writes.
    #[error("cannot {operation} from within a workflow")]
    InsideWorkflow {
        /// The operation that was called inside a workflow.
        operation: Cow<'static, str>,
    },

    /// An instance method was called from inside a workflow another instance is running.
    ///
    /// An instance method takes its executor from the handle it was called on, and its step ids
    /// from the ambient context. Normally those are the same instance and the distinction never
    /// surfaces. When they are not, there is no safe way to pick: the ids come from this
    /// workflow's counter and the checkpoint would be written through the other instance's system
    /// database, so it would land where the workflow that allocated them cannot see it — the
    /// replay skip would never match, the read would run again on recovery, and this workflow's
    /// counter would have moved on regardless.
    ///
    /// The free functions cannot raise this: they take both halves from the one context.
    #[error(
        "{operation} was called on a different DBOS instance than the one running this workflow"
    )]
    WrongInstance {
        /// The operation that was called on the wrong instance.
        operation: Cow<'static, str>,
    },

    /// The configuration could not be used.
    ///
    /// How the instance, the client or a queue was *set up* — a malformed database URL, an empty
    /// schema, a queue whose limits contradict each other. Not what a call was passed at the point
    /// it was made, which is [`InvalidArgument`](Self::InvalidArgument).
    #[error("invalid configuration: {0}")]
    Config(String),

    /// A call was given an argument it cannot act on.
    ///
    /// The caller's mistake at one call site, rather than a setting that was wrong before anything
    /// ran — waiting for the first of no workflows, or naming one forked id for a call that forks
    /// many. Kept apart from [`Config`](Self::Config) because the two are fixed in different
    /// places and by different people.
    ///
    /// **Deterministic, and recorded like any other failure** where the call took a step id: the
    /// arguments are not checkpointed, so a replay reads the refusal back rather than deciding it
    /// again against a set that may have changed.
    #[error("invalid argument to {operation}: {detail}")]
    InvalidArgument {
        /// The call that was given it.
        operation: Cow<'static, str>,
        /// What was wrong with it.
        detail: String,
    },

    /// Two workflows were registered under one identity.
    ///
    /// Uniqueness is on the whole `(name, class_name, config_name)` triple. A name resolving to
    /// two functions is a workflow that recovers as the wrong one, which is why this is an error
    /// rather than a last-registration-wins.
    #[error("a workflow is already registered as {workflow}")]
    AlreadyRegistered {
        /// The name that was registered twice.
        #[serde(alias = "key")]
        workflow: QualifiedWorkflowName,
    },

    /// A value could not be encoded for the database.
    ///
    /// A workflow's own `result` or `error` that cannot be encoded is that workflow's outcome: it
    /// is recorded as the workflow's error, and its handle's `result()` and every replay return
    /// it, rather than the body running again. A step's own `step result` or `step error` that
    /// cannot be encoded is likewise recorded as that step's error, so a replay returns it rather
    /// than running the step again.
    #[error("could not serialize the workflow {what}: {message}")]
    Serialization {
        /// Which value: `argument`, `result`, `error`.
        what: Cow<'static, str>,
        /// What the serializer said.
        message: String,
        /// The live failure, when this process is the one that produced it.
        ///
        /// `serde_json::Error` is not itself deserializable, so this is one of the two payloads
        /// here that cannot come back out of a column. [`message`](Self::Serialization::message)
        /// carries what it said, which is what a reader of a recorded error actually needs.
        #[serde(skip)]
        #[source]
        source: Option<serde_json::Error>,
    },

    /// A value from the database could not be decoded.
    ///
    /// Usually a signature that changed under a workflow already in flight — the row holds what
    /// the old code wrote, and the new code cannot read it.
    #[error("could not deserialize the workflow {what}: {message}")]
    Deserialization {
        /// Which value: `argument`, `result`, `error`.
        what: Cow<'static, str>,
        /// What the deserializer said.
        message: String,
        /// The live failure, when this process is the one that produced it.
        #[serde(skip)]
        #[source]
        source: Option<serde_json::Error>,
    },

    /// A row is tagged with an encoding this process cannot read.
    ///
    /// `rust_serde` and `portable_json` are always readable, and so is the configured
    /// [`CustomSerializer`](crate::CustomSerializer)'s name. Any other tag is another SDK's
    /// native encoding, or a custom serializer this process is not configured with, and a value
    /// in it is reported rather than guessed at.
    #[error("serialization {name} is not available")]
    UnknownSerialization {
        /// The tag the row carries.
        name: String,
    },

    /// A workflow was started that this executor has no registration for.
    ///
    /// In-process this cannot happen — a `WorkflowRef` comes from a registration. It is reachable
    /// through recovery and dequeue, where the name comes from a row and the code that registered
    /// it may be gone.
    #[error("no workflow is registered as {workflow}")]
    NotRegistered {
        /// The name that was looked up.
        #[serde(alias = "key")]
        workflow: QualifiedWorkflowName,
    },

    /// A workflow id resolved to no row at all.
    ///
    /// Unreachable through a handle this process minted — starting is what wrote the row — and
    /// reachable through anything that takes a caller's workflow id on faith.
    ///
    /// **[`status`](crate::WorkflowHandle::status) always reports it**: a single read has nothing
    /// to wait for. Awaiting the result reports it too when the handle came from a call that had
    /// the row in front of it, since a row that was there and is gone was deleted — and *waits*
    /// when the id is one nothing here has ever seen, which may name a workflow whose enqueue has
    /// not committed yet. The system database says the same thing as `NonExistentWorkflow` and this
    /// is what a caller sees instead, so an unchecked id is one error rather than two shapes of the
    /// same error.
    #[error("no workflow exists with id {workflow_id}")]
    WorkflowNotFound {
        /// The id that matched nothing.
        workflow_id: String,
    },

    /// The workflow was cancelled by shutdown while this caller was waiting for it.
    ///
    /// Its row stays `PENDING`, so a later executor recovers it. Nothing was lost; this caller
    /// simply stopped being the one waiting.
    #[error("the workflow {workflow_id} was interrupted by shutdown and left PENDING")]
    Interrupted {
        /// The workflow that was interrupted.
        workflow_id: String,
    },

    /// A workflow failed, and what it failed with did not decode.
    ///
    /// The workflow-level twin of [`StepFailed`](Self::StepFailed). The row's tag names an
    /// encoding this process reads, but the error in it is not one it can rebuild: written by
    /// another build whose error type has since changed shape, or a malformed portable envelope.
    /// The stored text is what survives. A workflow error that decodes comes back as the error
    /// itself, and a portable one as [`Portable`](Self::Portable).
    #[error("the workflow {workflow_id} failed: {message}")]
    WorkflowFailed {
        /// The workflow that failed.
        workflow_id: String,
        /// What it reported.
        message: String,
    },

    /// A workflow failed with an error stored portably: `{name, message, code, data}`.
    ///
    /// The shape every DBOS SDK writes a portable workflow's error in, so this is how an error
    /// raised in another language arrives. A portable Rust workflow's own error does not: its
    /// envelope carries the error's fields as `data`, and a reader whose error type it is gets the
    /// error itself back — [`Application`](Self::Application), or the DBOS variant it was. This is
    /// what is left when the envelope names a type the reader does not have.
    ///
    /// `name` is the writer's name for the error's type, which is not spelt the same by every
    /// writer — a type name, a symbolic error code — so match on it knowing who wrote the error.
    /// `code` and `data` are whatever the writer put there, often nothing.
    #[error("{name}: {message}")]
    Portable {
        /// What kind of error it was, in the writer's terms.
        name: String,
        /// What it said.
        message: String,
        /// The writer's error code, if it gave one: a number or a string.
        code: Option<serde_json::Value>,
        /// Anything else the writer attached.
        data: Option<serde_json::Value>,
    },

    /// The workflow was cancelled.
    ///
    /// Inside a running workflow, this is what the call that notices the cancel returns: a step in
    /// progress when its result can't be recorded, a later call that checkpoints, and a preemptible
    /// step at once. [`join_workflows`](fn@crate::join_workflows) records nothing, so it doesn't notice
    /// and goes on waiting. Return the error with `?`, after any clean-up the workflow wants to do.
    #[error("the workflow {workflow_id} was cancelled")]
    WorkflowCancelled {
        /// The workflow that was cancelled.
        workflow_id: String,
    },

    /// A workflow this one was **waiting on** was cancelled.
    ///
    /// Not [`WorkflowCancelled`](Self::WorkflowCancelled), and the distinction is the whole reason
    /// this variant exists: that one means *this* workflow is being cancelled and must stop, while
    /// this one means the workflow it awaited will never produce a result. A parent told the first
    /// when its child was cancelled would stop itself, which is not what happened: the awaiting
    /// workflow is not being cancelled.
    ///
    /// It is the awaited workflow's *outcome*, so it is recorded like one: a replayed parent is
    /// told the same thing without waiting again.
    #[error("the workflow {workflow_id} this one was awaiting was cancelled")]
    AwaitedWorkflowCancelled {
        /// The awaited workflow that was cancelled — the child, not the caller.
        workflow_id: String,
    },

    /// The workflow was recovered too many times and is parked.
    #[error("the workflow {workflow_id} exceeded {recovery_attempts} recovery attempts")]
    MaxRecoveryAttemptsExceeded {
        /// The workflow that was parked.
        workflow_id: String,
        /// How many attempts it took.
        recovery_attempts: i64,
    },

    /// A recorded step failed, and what it failed with was not one of ours to decode.
    ///
    /// Raised on replay against a row another SDK wrote, whose serializer chose its own shape. The
    /// stored text is what survives. A step error that decodes replays as the error itself.
    #[error("the step {step} failed: {message}")]
    StepFailed {
        /// The step's name.
        step: String,
        /// What it reported when it ran.
        message: String,
    },

    /// A step was polled somewhere its step id cannot be honoured.
    ///
    /// A step takes its id from the workflow's counter **where it is built**, which is a claim on
    /// one position in one workflow. So there are only four places a built step may be polled,
    /// and **everything else is refused** rather than run under an id nothing there can honour:
    ///
    /// - at a step boundary of the workflow it was built in, which is the ordinary case and the
    ///   only one that records anything;
    /// - inside the very step body it was built in, where the leaf rule makes it a plain call;
    /// - outside a workflow, if that is also where it was built, which is what makes a function
    ///   built from steps ordinarily testable;
    /// - anywhere at all, if it was reached through a [`Client`](crate::Client)'s connection: a
    ///   client has no step counter anywhere to disagree with, so its calls are pinned to nothing.
    ///
    /// The refusals worth naming, because they are the ones code reaches by accident:
    ///
    /// - **Built outside a workflow, awaited inside one.** It took no id, so it would run
    ///   unrecorded where the surrounding workflow plainly expects a checkpoint, and every replay
    ///   would run it again. Easy to reach by accident: a step built in the handler that then
    ///   starts the workflow, and awaited from the workflow body.
    /// - **Built in one workflow, polled in another.** Its row would land under the wrong
    ///   workflow's id.
    /// - **Built inside a step body, awaited in the workflow proper**, or inside a *different*
    ///   body. By the leaf rule it took no id inside the body it was built in, so it would run
    ///   unrecorded where a checkpoint was expected.
    /// - **Built in the workflow proper, carried into a step body.** Its checkpoint would sit
    ///   beneath a step whose own row already stands for whatever its body did.
    /// - **Polled where there is no context at all**, which in practice means
    ///   `tokio::spawn(step(..))` or `smol::spawn(step(..))`. A spawned task inherits no workflow,
    ///   so the step would run undurably wherever it was built — and a step built in the workflow
    ///   proper has already spent an id that nothing will ever record against. Await the step in
    ///   the workflow, or spawn work that is not a step.
    ///
    /// `built` and `polled` name the two places, so a refusal reads "built inside a step of
    /// workflow w but polled in workflow w" rather than naming the same place twice.
    ///
    /// **Not a control signal.** It is a mistake in the code rather than a failure of the engine's
    /// substrate, it is deterministic, and a replay reaches it again — so it is the workflow's
    /// outcome like any other failure.
    #[error(
        "step {step} was built {built} but polled {polled}: a step takes its id where it is built"
    )]
    StepBuiltElsewhere {
        /// The step's name, which is the only thing a caller can find it by.
        step: String,
        /// Where its id came from, or that it has none.
        built: std::borrow::Cow<'static, str>,
        /// Where it was polled instead.
        polled: std::borrow::Cow<'static, str>,
    },

    /// Another durable call of the same workflow took the step position a
    /// [`patch`](crate::patch()) or [`deprecate_patch`](crate::deprecate_patch()) was built at.
    ///
    /// Either call fixes its position where it is built, as every durable call does, but takes it
    /// only once it has read the workflow's history there, when awaited. A call built in between —
    /// written after the patch and awaited before it, or built in another branch of the same body,
    /// one arm of a `tokio::join!` — took the position first. Refused rather than recorded
    /// elsewhere: await a patch where it is written, from sequential workflow code, before any
    /// branches start or after they finish.
    ///
    /// **A mistake in the code, and recorded like one.** It is the workflow's outcome, as
    /// [`StepBuiltElsewhere`](Self::StepBuiltElsewhere) is. A call built between a patch and its
    /// await is refused on every execution, and so is a branch whose call is built before the
    /// patch is first awaited.
    #[error(
        "patch {patch} in workflow {workflow_id} lost step {step_id} to a call built before it \
         was awaited: await a patch where it is written, from sequential workflow code"
    )]
    PatchNondeterminism {
        /// The workflow that was patching.
        workflow_id: String,
        /// The patch's name, as the code gave it.
        patch: String,
        /// The position the patch looked at, which another call took.
        step_id: i32,
    },

    /// A step attempt ran past its timeout and was stopped.
    ///
    /// **Ordinary failure, not a control signal**: it is offered to the retry predicate and
    /// retried like any other. A step whose attempts all time out fails with
    /// [`MaxStepRetriesExceeded`](Self::MaxStepRetriesExceeded) holding one of these per attempt.
    ///
    /// The timeout bounds **one attempt**, not the step: three attempts at five seconds may take
    /// fifteen seconds of body time, plus backoff. Nothing here carries a numeric error code,
    /// because the portable error shape matches on `name` and makes `code` optional.
    #[error("the step {step} exceeded its {}ms timeout", timeout.as_millis())]
    StepTimeout {
        /// The step's name.
        step: String,
        /// The timeout it exceeded.
        timeout: std::time::Duration,
    },

    /// A stream read waited out its timeout with no value, or the stream ended before the offset
    /// a single-value read asked for.
    ///
    /// The timeout is **per value**: the clock restarts each time the reader delivers one. A
    /// workflow's read records this as the step's outcome, so a replay raises it again without
    /// waiting — the same way a timed-out `get_event` replays its `None`.
    ///
    /// `timeout` is `None` for the single-value read of a stream that ended first, where no value
    /// will ever arrive and there was no wait to time out.
    #[error(
        "no value arrived on stream {key} of workflow {workflow_id}{}",
        timeout.map_or_else(String::new, |t| format!(" within {}ms", t.as_millis()))
    )]
    StreamTimeout {
        /// The workflow whose stream was being read.
        workflow_id: String,
        /// The stream's key.
        key: String,
        /// The per-value timeout that elapsed, or `None` if the stream ended first.
        timeout: Option<std::time::Duration>,
    },

    /// A step was retried to its limit and every attempt failed.
    ///
    /// Carries **all** of them rather than the last. The first failure is usually the informative
    /// one and the last is usually a timeout, so keeping only one loses the half that explains the
    /// other.
    ///
    /// Raised only where retries were actually asked for. A step left at the default
    /// [`StepOptions::max_attempts`](crate::StepOptions::max_attempts) of 1 records whatever its one
    /// attempt failed with, unwrapped: there is no retry policy to report on, and wrapping would
    /// make every ordinary step failure arrive inside a collection of one.
    #[error("the step {step} failed on all {attempts} attempts")]
    MaxStepRetriesExceeded {
        /// The step's name.
        step: String,
        /// How many attempts were made.
        attempts: u32,
        /// What each attempt failed with, oldest first.
        errors: Vec<Error<E>>,
    },

    /// The system database failed.
    ///
    /// Deliberately not a `#[from]`: that impl overlaps the blanket `From<E> for Error<E>` at
    /// `Error<sysdb::Error>`. Engine code names the variant instead.
    #[error(transparent)]
    SystemDatabase(crate::sysdb::Error),
}

/// The conversion that makes `?` work on an application's own error.
///
/// Accepted only because `Error` is generic — see the module documentation.
impl<E> From<E> for Error<E> {
    fn from(error: E) -> Self {
        Error::Application(error)
    }
}

/// A system-database refusal's field and detail as one phrase, for a message that names the call
/// separately: "`delay` must be a positive, non-zero duration".
pub(crate) fn refusal_detail(field: &str, detail: &str) -> String {
    format!("`{field}` {detail}")
}

impl<E> Error<E> {
    /// A system-database failure, in the engine's terms.
    ///
    /// An unknown tag on a checkpoint the system database read back through this process's
    /// serializer is [`UnknownSerialization`](Self::UnknownSerialization), exactly as it is on any
    /// other checkpoint, so a caller matching on it sees one variant and a workflow that hits it
    /// fails the same way wherever the row was. Everything else goes through
    /// [`from_execution`](Self::from_execution).
    pub(crate) fn from_sysdb(error: crate::sysdb::Error) -> Self {
        match error {
            crate::sysdb::Error::UnknownSerialization { name } => {
                Error::UnknownSerialization { name }
            }
            other => Self::from_execution(other),
        }
    }

    /// A system-database failure met by a workflow's own execution.
    ///
    /// Cancellation is [`WorkflowCancelled`](Self::WorkflowCancelled), so code in a workflow matches
    /// one variant whichever call noticed the cancel, and can't mistake it for the
    /// [`SystemDatabase`](Self::SystemDatabase) failures it would otherwise sit among. Everything
    /// else stays [`SystemDatabase`](Self::SystemDatabase).
    pub(crate) fn from_execution(error: crate::sysdb::Error) -> Self {
        match error {
            crate::sysdb::Error::WorkflowCancelled { workflow_id } => {
                Error::WorkflowCancelled { workflow_id }
            }
            other => Error::SystemDatabase(other),
        }
    }

    /// A system-database refusal of one of a call's arguments, as the surface reports it.
    ///
    /// `refuse` builds the error from [`InvalidInput`](crate::sysdb::Error::InvalidInput)'s field
    /// and detail, which lets each surface name the field the way its own API spells it. Any other
    /// error is not a refusal and goes through [`from_sysdb`](Self::from_sysdb), so a
    /// backend or connection failure keeps its [`SystemDatabase`](Self::SystemDatabase) variant.
    ///
    /// For a surface that refuses an argument by asking the system database's own check rather
    /// than restating it: the two then cannot disagree about what is storable, and the system
    /// database still refuses the same value for callers that reach it directly.
    pub(crate) fn from_refusal(
        error: crate::sysdb::Error,
        refuse: impl FnOnce(&str, String) -> Self,
    ) -> Self {
        match error {
            crate::sysdb::Error::InvalidInput { field, detail } => refuse(&field, detail),
            other => Self::from_sysdb(other),
        }
    }

    /// [`from_refusal`](Self::from_refusal) as an [`InvalidArgument`](Self::InvalidArgument) to
    /// `operation`, its detail phrased by [`refusal_detail`]: "`duration` must fit in milliseconds
    /// as a 64-bit integer".
    pub(crate) fn refused_argument(operation: &'static str, error: crate::sysdb::Error) -> Self {
        Self::from_refusal(error, |field, detail| Error::InvalidArgument {
            operation: operation.into(),
            detail: refusal_detail(field, &detail),
        })
    }

    /// A value that could not be encoded for its column; `what` names it, as `"argument"` or
    /// `"result"` does.
    pub(crate) fn encode_failed(what: &'static str, source: serde_json::Error) -> Self {
        Error::Serialization {
            what: what.into(),
            message: source.to_string(),
            source: Some(source),
        }
    }

    /// A value that could not be decoded from its column; `what` names it, as `"argument"` or
    /// `"result"` does.
    pub(crate) fn decode_failed(what: &'static str, source: serde_json::Error) -> Self {
        Error::Deserialization {
            what: what.into(),
            message: source.to_string(),
            source: Some(source),
        }
    }

    /// Re-targets this error at another application error type.
    ///
    /// One match over the engine's variants, kept in a single place so [`lift`](Self::lift) and
    /// [`PendingStep::map_error`](crate::PendingStep::map_error) share the list rather than each
    /// carrying a copy of it.
    /// `Fn + Copy` rather than `FnOnce` because
    /// [`MaxStepRetriesExceeded`](Self::MaxStepRetriesExceeded) nests errors and so recurses; the
    /// closures handed to it are trivially `Copy`, so the tighter bound costs nothing.
    pub(crate) fn map_application<E2>(self, f: impl Fn(E) -> E2 + Copy) -> Error<E2> {
        match self {
            Error::Application(error) => Error::Application(f(error)),
            Error::NotLaunched { operation } => Error::NotLaunched { operation },
            Error::AlreadyLaunched { operation } => Error::AlreadyLaunched { operation },
            Error::NotInWorkflow { operation } => Error::NotInWorkflow { operation },
            Error::InsideStep { operation } => Error::InsideStep { operation },
            Error::InsideWorkflow { operation } => Error::InsideWorkflow { operation },
            Error::WrongInstance { operation } => Error::WrongInstance { operation },
            Error::Config(message) => Error::Config(message),
            Error::InvalidArgument { operation, detail } => {
                Error::InvalidArgument { operation, detail }
            }
            Error::AlreadyRegistered { workflow } => Error::AlreadyRegistered { workflow },
            Error::Serialization {
                what,
                message,
                source,
            } => Error::Serialization {
                what,
                message,
                source,
            },
            Error::Deserialization {
                what,
                message,
                source,
            } => Error::Deserialization {
                what,
                message,
                source,
            },
            Error::UnknownSerialization { name } => Error::UnknownSerialization { name },
            Error::NotRegistered { workflow } => Error::NotRegistered { workflow },
            Error::WorkflowNotFound { workflow_id } => Error::WorkflowNotFound { workflow_id },
            Error::Interrupted { workflow_id } => Error::Interrupted { workflow_id },
            Error::WorkflowFailed {
                workflow_id,
                message,
            } => Error::WorkflowFailed {
                workflow_id,
                message,
            },
            Error::Portable {
                name,
                message,
                code,
                data,
            } => Error::Portable {
                name,
                message,
                code,
                data,
            },
            Error::WorkflowCancelled { workflow_id } => Error::WorkflowCancelled { workflow_id },
            Error::AwaitedWorkflowCancelled { workflow_id } => {
                Error::AwaitedWorkflowCancelled { workflow_id }
            }
            Error::MaxRecoveryAttemptsExceeded {
                workflow_id,
                recovery_attempts,
            } => Error::MaxRecoveryAttemptsExceeded {
                workflow_id,
                recovery_attempts,
            },
            Error::StepFailed { step, message } => Error::StepFailed { step, message },
            Error::StepTimeout { step, timeout } => Error::StepTimeout { step, timeout },
            Error::StreamTimeout {
                workflow_id,
                key,
                timeout,
            } => Error::StreamTimeout {
                workflow_id,
                key,
                timeout,
            },
            Error::StepBuiltElsewhere {
                step,
                built,
                polled,
            } => Error::StepBuiltElsewhere {
                step,
                built,
                polled,
            },
            Error::PatchNondeterminism {
                workflow_id,
                patch,
                step_id,
            } => Error::PatchNondeterminism {
                workflow_id,
                patch,
                step_id,
            },
            Error::MaxStepRetriesExceeded {
                step,
                attempts,
                errors,
            } => Error::MaxStepRetriesExceeded {
                step,
                attempts,
                errors: errors
                    .into_iter()
                    .map(|error| error.map_application(f))
                    .collect(),
            },
            Error::SystemDatabase(error) => Error::SystemDatabase(error),
        }
    }

    /// The control signal this failure is, if it is one.
    ///
    /// A control error is not the workflow's *outcome*: a cancelled workflow recorded as having
    /// failed would come back permanently failed, having lost that it was interrupted rather than
    /// wrong. So the recording layer asks this before writing anything terminal.
    ///
    /// A plain match on the outermost variant is enough. That is a consequence of the inversion:
    /// an application error type has nowhere to hide one of ours, so a control signal is either
    /// the outermost variant or it is not present.
    pub(crate) fn control(&self) -> Option<Error<EngineOnly>> {
        match self {
            Error::WorkflowCancelled { workflow_id } => Some(Error::WorkflowCancelled {
                workflow_id: workflow_id.clone(),
            }),
            Error::Interrupted { workflow_id } => Some(Error::Interrupted {
                workflow_id: workflow_id.clone(),
            }),
            // Every system-database failure but a typed refusal. A database failure is a failure
            // of the engine's substrate; it is never a statement about what the workflow
            // computed, so recording it as that workflow's outcome asserts something false. A
            // transient blip while the engine checkpoints would otherwise permanently fail a
            // workflow that had not failed — and the row is the only copy of that fact.
            //
            // A deterministic one — `UnexpectedStep`, `Malformed` — retries rather than failing
            // fast. That is the deliberate trade: `MAX_RECOVERY_ATTEMPTS` bounds it and parks the
            // workflow, where recording ERROR would have discarded work a fixed deployment could
            // still have finished.
            //
            // A typed refusal — "already exists", "no such workflow" — is the opposite case: an
            // answer the call gave, which its step records and every replay gives back. Waiting
            // for a later recovery cannot change it, so a workflow that returns one has failed
            // with it, as it would with any step's error.
            Error::SystemDatabase(error) if !error.should_record() => {
                Some(Error::SystemDatabase(error.clone()))
            }
            _ => None,
        }
    }
}

impl Error<EngineOnly> {
    /// Carries an engine error into a durable function's own error channel.
    ///
    /// Total, and that is the point of [`EngineOnly`] being uninhabited: the `Application` arm
    /// holds a value of a type with no values, so the compiler discharges it rather than this
    /// needing a panic or a fallback for a case that cannot occur.
    ///
    /// Public because a workflow with its own error type sometimes holds an engine-channel
    /// result — the management methods on [`DBOS`](crate::DBOS) are, as are the bulk free
    /// [`resume_workflows`](crate::resume_workflows) and [`fork_workflows`](crate::fork_workflows), and
    /// [`DBOS::get_event`](crate::DBOS::get_event) called from inside a workflow is another — and
    /// `?` cannot lift `Error<EngineOnly>` into `Error<E>` on its own: the blanket conversion
    /// would wrap the whole error as the application's. This is the conversion written out:
    ///
    /// ```ignore
    /// let status = child.status().await.map_err(Error::lift)?;
    /// ```
    ///
    /// The workflow's own calls do not need it: [`step`](crate::step()),
    /// [`set_event`](crate::set_event) and [`get_event`](crate::get_event) answer in the caller's
    /// channel, so `?` works on them directly. A child's [`start`](crate::WorkflowRef::start),
    /// [`run`](crate::WorkflowRef::run) and the await of its handle answer in the *child's*
    /// channel, which is the same channel whenever parent and child fail the same way — the common
    /// case, and the one those calls read as written for.
    ///
    /// Where the two differ, only the start is this conversion's business: it can fail in nothing
    /// but the engine's terms, so [`PendingStart::lift`](crate::PendingStart::lift) re-declares
    /// which channel it answers in, before the await rather than converting after it. A `run` or a
    /// handle's await carries the child's own application failure, which no lift can reach: that
    /// one is a real error of the child's type, and the parent writes out what it makes of it.
    pub fn lift<E>(self) -> Error<E> {
        self.map_application(|impossible| match impossible {})
    }
}

/// How an erased workflow failed.
///
/// Two cases, and keeping them apart is the point: what the workflow itself returned, which is its
/// outcome and gets recorded, and a signal from the engine, which is not. A cancelled workflow
/// recorded as having *failed* would come back permanently failed, having lost that it was
/// interrupted rather than wrong.
#[derive(Debug)]
pub(crate) enum Failure {
    /// The workflow's own failure, encoded as the workflow's row records it. This is what the
    /// error column holds and what a replay decodes.
    Recorded(EncodedError),
    /// One of ours. Never recorded as the workflow's outcome.
    Control(Error<EngineOnly>),
}

/// What a row records as a workflow's or a step's error when its custom serializer refuses even the
/// stand-in [`EncodedError::refused`] offers: fixed text, so nothing of the error itself is stored
/// unencoded.
const UNENCODABLE_ERROR: &str = "the error could not be encoded by its serializer";

/// A workflow's or a step's error as its row holds it, with the encoding that says how to read it.
///
/// **The encoding travels with the bytes** because the encodings write an error differently — the
/// whole [`Error`] natively or in a custom serializer, a `{name, message, code, data}` envelope
/// portably — and a reader that guessed would read one as the other.
#[derive(Debug, Clone)]
pub(crate) struct EncodedError {
    /// The encoded error.
    pub(crate) encoded: String,
    /// How it is encoded.
    pub(crate) encoding: PayloadEncoding,
}

impl EncodedError {
    /// `error` as its workflow's row records it, in `encoding`.
    ///
    /// **Total**, because the row has to end. A failure that is not recorded leaves the row
    /// `PENDING`, and recovery would run the whole body again, side effects and all, only to fail
    /// to encode the same error the same way. An error the encoding refuses is recorded as
    /// [`refused`](Self::refused) describes.
    pub(crate) fn of<E: DurableError>(error: &Error<E>, encoding: PayloadEncoding) -> Self {
        match encode_error::<E, EngineOnly>(error, &encoding) {
            Ok(encoded) => Self { encoded, encoding },
            Err(refused) => Self::refused(error, refused, encoding, "error"),
        }
    }

    /// `error`, which `encoding` has refused with `refused`, as its row records it instead.
    ///
    /// Recorded as [`Error::Serialization`] of `what` — `error` for a workflow's, `step error` for
    /// a step's — with what the serializer said and what the error displayed as its message. That
    /// one is strings alone, which the built-in encodings always write. For a custom serializer
    /// that refuses it too, the column holds [`UNENCODABLE_ERROR`], a fixed marker, which a reader
    /// reports as an error that does not decode. **Never the error's own details**: a custom
    /// serializer may exist to keep them out of the database, an encrypting one for instance, and
    /// text the caller chose could also hold a byte the column cannot store. The details go to the
    /// log instead.
    pub(crate) fn refused<E: DurableError>(
        error: &Error<E>,
        refused: Error<EngineOnly>,
        encoding: PayloadEncoding,
        what: &'static str,
    ) -> Self {
        let said = match refused {
            Error::Serialization { message, .. } => message,
            other => other.to_string(),
        };
        let stand_in = Error::<E>::Serialization {
            what: what.into(),
            message: format!("{said}; the error was: {error}"),
            source: None,
        };
        tracing::warn!(
            what,
            error = %stand_in,
            "an error could not be encoded; its message is recorded instead"
        );
        let encoded = encode_error::<E, EngineOnly>(&stand_in, &encoding).unwrap_or_else(|_| {
            tracing::warn!(
                what,
                "the serializer refused the stand-in too; a fixed marker is recorded"
            );
            UNENCODABLE_ERROR.to_owned()
        });
        Self { encoded, encoding }
    }

    /// The error this records, in the caller's error type.
    ///
    /// An error that does not decode — a native one some other build wrote in a shape this one
    /// cannot read — is reported as [`Error::WorkflowFailed`] with the bytes as its message, which
    /// is what survives. One a custom serializer cannot decode is [`Error::Deserialization`]: its
    /// bytes are no message.
    pub(crate) fn decode<E: DurableError>(&self, workflow_id: &str) -> Error<E> {
        decode_error(&self.encoded, &self.encoding).unwrap_or_else(|| Error::WorkflowFailed {
            workflow_id: workflow_id.to_owned(),
            message: self.encoded.clone(),
        })
    }
}

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

    #[derive(Debug, thiserror::Error, serde::Serialize, serde::Deserialize, PartialEq)]
    #[error("the card was declined after {attempts} attempts")]
    struct CardDeclined {
        attempts: u32,
    }

    /// The property the type parameter exists for: an application error is held as itself, so it
    /// comes back as itself.
    #[test]
    fn an_application_error_round_trips_whole() {
        let error: Error<CardDeclined> = CardDeclined { attempts: 3 }.into();
        let json = serde_json::to_string(&error).unwrap();
        let back: Error<CardDeclined> = serde_json::from_str(&json).unwrap();

        let Error::Application(back) = back else {
            panic!("expected an application error, got {back:?}")
        };
        assert_eq!(back, CardDeclined { attempts: 3 }, "the field survives");
        assert_eq!(error.to_string(), back.to_string());
    }

    /// An error checkpointed before its workflow field was named `workflow` still replays.
    #[test]
    fn a_registration_error_recorded_as_key_still_decodes() {
        for (json, expected) in [
            (
                r#"{"NotRegistered":{"key":{"name":"f","class_name":"C","config_name":null}}}"#,
                "no workflow is registered as f/C",
            ),
            (
                r#"{"AlreadyRegistered":{"key":{"name":"f","class_name":null,"config_name":null}}}"#,
                "a workflow is already registered as f",
            ),
        ] {
            let back: Error = serde_json::from_str(json).unwrap();
            assert_eq!(back.to_string(), expected);
        }
    }

    /// `?` on the application's own error needs no impl from the application.
    #[test]
    fn the_blanket_conversion_lifts_an_application_error() {
        fn fallible() -> Result<(), CardDeclined> {
            Err(CardDeclined { attempts: 1 })?
        }
        assert!(matches!(
            fallible().unwrap_err(),
            Error::Application(CardDeclined { attempts: 1 })
        ));
    }

    #[test]
    fn an_engine_error_lifts_into_any_channel() {
        let engine: Error = Error::NotLaunched {
            operation: "run a workflow".into(),
        };
        let lifted: Error<CardDeclined> = engine.lift();
        assert_eq!(
            lifted.to_string(),
            "cannot run a workflow before DBOS is launched"
        );
    }

    #[test]
    fn an_execution_sees_cancellation_at_the_top_level() {
        let cancelled = crate::sysdb::Error::WorkflowCancelled {
            workflow_id: "wf-1".to_owned(),
        };
        assert!(matches!(
            Error::<CardDeclined>::from_execution(cancelled.clone()),
            Error::WorkflowCancelled { workflow_id } if workflow_id == "wf-1"
        ));
        assert!(matches!(
            Error::<CardDeclined>::from_sysdb(cancelled),
            Error::WorkflowCancelled { workflow_id } if workflow_id == "wf-1"
        ));

        let conflict = crate::sysdb::Error::WorkflowConflict {
            workflow_id: "wf-1".to_owned(),
        };
        assert!(matches!(
            Error::<CardDeclined>::from_execution(conflict),
            Error::SystemDatabase(crate::sysdb::Error::WorkflowConflict { .. })
        ));
    }

    #[test]
    fn only_control_errors_report_themselves_as_control() {
        let cancelled: Error<CardDeclined> = Error::WorkflowCancelled {
            workflow_id: "wf-1".to_owned(),
        };
        assert!(cancelled.control().is_some());

        let failed: Error<CardDeclined> = CardDeclined { attempts: 1 }.into();
        assert!(
            failed.control().is_none(),
            "an application failure is an outcome, not a signal"
        );
    }

    #[test]
    fn a_typed_refusal_is_an_outcome_and_database_trouble_a_signal() {
        let refused: Error = Error::SystemDatabase(crate::sysdb::Error::AlreadyRegistered {
            kind: "Schedule".into(),
            name: "nightly".to_owned(),
        });
        assert!(
            refused.control().is_none(),
            "a refusal is the call's answer, so a workflow returning it has failed"
        );

        let signals: [crate::sysdb::Error; 4] = [
            crate::sysdb::Error::Backend(crate::sysdb::BackendError {
                message: "relation does not exist".to_owned(),
                sqlstate: Some("42P01".to_owned()),
                kind: crate::sysdb::BackendErrorKind::Permanent,
            }),
            crate::sysdb::Error::WorkflowCancelled {
                workflow_id: "wf-1".to_owned(),
            },
            crate::sysdb::Error::StepNondeterminism {
                workflow_id: "wf-1".to_owned(),
                step_id: 0,
            },
            crate::sysdb::Error::Malformed("unreadable".to_owned()),
        ];
        for signal in signals {
            let error: Error = Error::SystemDatabase(signal);
            assert!(error.control().is_some(), "{error} is a signal");
        }
    }

    /// An application error that cannot be written as JSON: `serde_json` has no spelling for a
    /// non-string map key.
    #[derive(Debug, thiserror::Error, serde::Serialize, serde::Deserialize)]
    #[error("no keys fit")]
    struct BadKeys {
        by_pair: std::collections::HashMap<(u8, u8), u8>,
    }

    fn bad_keys() -> Error<BadKeys> {
        BadKeys {
            by_pair: [((1, 2), 3)].into_iter().collect(),
        }
        .into()
    }

    /// An error its encoding refuses is still recorded, in that encoding, as the refusal: what the
    /// serializer said and what the error displayed, rather than nothing at all.
    #[test]
    fn an_error_its_encoding_refuses_is_recorded_as_the_refusal() {
        for encoding in [PayloadEncoding::Native, PayloadEncoding::Portable] {
            let recorded = EncodedError::of(&bad_keys(), encoding.clone());
            assert_eq!(recorded.encoding, encoding);
            let read = recorded.decode::<BadKeys>("wf-1");
            let Error::Serialization { what, message, .. } = &read else {
                panic!("{} read back as {read:?}", encoding.name())
            };
            assert_eq!(what, "error", "{}", encoding.name());
            assert!(
                message.contains("key must be a string") && message.ends_with("no keys fit"),
                "{}: {message}",
                encoding.name()
            );
        }
    }

    /// A step's refused error is recorded the same way, as a refusal of the step's error, and a
    /// [`Error::MaxStepRetriesExceeded`] holding refused attempts is refused as a whole.
    #[test]
    fn a_steps_refused_error_is_recorded_as_a_refusal_of_the_step_error() {
        let retried: Error<BadKeys> = Error::MaxStepRetriesExceeded {
            step: "charge".to_owned(),
            attempts: 2,
            errors: vec![bad_keys(), bad_keys()],
        };
        for error in [bad_keys(), retried] {
            for encoding in [PayloadEncoding::Native, PayloadEncoding::Portable] {
                let refused = encode_error::<BadKeys, EngineOnly>(&error, &encoding)
                    .expect_err("the encoding refuses a non-string key");
                let recorded =
                    EncodedError::refused(&error, refused, encoding.clone(), "step error");
                let read = recorded.decode::<BadKeys>("wf-1");
                let Error::Serialization { what, message, .. } = &read else {
                    panic!("{} read back as {read:?}", encoding.name())
                };
                assert_eq!(what, "step error", "{}", encoding.name());
                assert!(
                    message.contains("key must be a string")
                        && message.ends_with(&error.to_string()),
                    "{}: {message}",
                    encoding.name()
                );
            }
        }
    }

    /// A custom serializer that refuses even the stand-in leaves a fixed marker, so the row still
    /// ends without anything of the error stored unencoded.
    #[test]
    fn a_serializer_that_refuses_everything_records_a_fixed_marker() {
        use crate::serialization::{CustomSerializer, Payload, PayloadSlot, Serializer};

        #[derive(Debug)]
        struct Refuses;

        impl CustomSerializer for Refuses {
            fn name(&self) -> &str {
                "refuses"
            }

            fn serialize(
                &self,
                _: Payload<'_>,
            ) -> std::result::Result<String, Box<dyn std::error::Error + Send + Sync>> {
                Err("nothing goes".into())
            }

            fn deserialize(
                &self,
                _: &str,
                _: PayloadSlot<'_>,
            ) -> std::result::Result<(), Box<dyn std::error::Error + Send + Sync>> {
                Err("nothing comes back".into())
            }
        }

        let error: Error<CardDeclined> = CardDeclined { attempts: 3 }.into();
        let recorded = EncodedError::of(&error, Serializer::custom(Refuses).encoding());
        // Nothing of the error reaches the column: the serializer may be there to keep it out.
        assert_eq!(recorded.encoded, UNENCODABLE_ERROR);
        // Text the serializer cannot decode back, which a reader reports as that.
        let read = recorded.decode::<CardDeclined>("wf-1");
        assert!(
            matches!(&read, Error::Deserialization { what, .. } if what == "error"),
            "{read:?}"
        );
    }

    /// A custom serializer whose text always holds a NUL byte, which a text column cannot store, is
    /// refused as one that refuses everything: the error and its stand-in alike, so the row records
    /// the fixed marker instead of failing to write.
    #[test]
    fn a_serializer_that_writes_nul_bytes_records_a_fixed_marker() {
        use crate::serialization::{CustomSerializer, Payload, PayloadSlot, Serializer};

        #[derive(Debug)]
        struct WritesNul;

        impl CustomSerializer for WritesNul {
            fn name(&self) -> &str {
                "writes_nul"
            }

            fn serialize(
                &self,
                value: Payload<'_>,
            ) -> std::result::Result<String, Box<dyn std::error::Error + Send + Sync>> {
                Ok(format!("{}\0", serde_json::to_string(&value)?))
            }

            fn deserialize(
                &self,
                _: &str,
                _: PayloadSlot<'_>,
            ) -> std::result::Result<(), Box<dyn std::error::Error + Send + Sync>> {
                Err("nothing comes back".into())
            }
        }

        let encoding = Serializer::custom(WritesNul).encoding();
        let error: Error<CardDeclined> = CardDeclined { attempts: 3 }.into();
        let refused = encode_error::<CardDeclined, EngineOnly>(&error, &encoding)
            .expect_err("a NUL byte is refused");
        assert!(
            matches!(&refused, Error::Serialization { what, message, .. }
                if what == "error" && message.contains("NUL")),
            "{refused:?}"
        );
        let recorded = EncodedError::of(&error, encoding);
        assert_eq!(recorded.encoded, UNENCODABLE_ERROR);
    }
}