dbos 0.6.0

//! What can go wrong talking to the system database.
//!
//! Split from the trait itself because the error surface is shared: the retry layer classifies
//! on it, both input types validate into it, and a second backend will construct it without
//! seeing any of the trait's method signatures.

use std::borrow::Cow;

use super::types::WorkflowStatus;

/// What went wrong talking to the system database.
///
/// Deliberately not `sqlx::Error`: that type names a specific driver, and
/// [`SystemDatabase`](super::SystemDatabase) has to be implementable by a backend that does not
/// use one.
///
/// Serializable, because the engine records a failed step's error and has to give back *that
/// error* on replay rather than a description of it. Nothing here names a driver type, so this
/// costs a derive: the same rule that keeps a second backend possible is what makes it encodable.
/// The `&'static str` discriminants are [`Cow`] for the same reason — a borrowed
/// string has nothing to borrow from when it comes back out of a column.
#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub enum Error {
    /// The database rejected or could not serve the request.
    Backend(BackendError),
    /// A stored value could not be understood — an unrecognised status, a missing column.
    ///
    /// Usually means the database was written by an implementation that knows something this
    /// one does not.
    Malformed(String),
    /// A recorded value carries a `serialization` tag the caller's
    /// [`StepCodec`](super::types::StepCodec) cannot read.
    ///
    /// Reported rather than guessed at: decoding bytes as an encoding they were not written in
    /// would hand a replay a value the first run never saw.
    UnknownSerialization {
        /// The tag the row carries.
        name: String,
    },
    /// A workflow with this id already exists, running a different function.
    ///
    /// Reusing an id for different work is a programming error, not a race: the id is how
    /// two attempts are recognised as the same workflow.
    ConflictingWorkflow {
        /// The id submitted twice.
        workflow_id: String,
        /// Which part disagreed, and how.
        detail: String,
    },
    /// A workflow with this id already exists, and the caller asked for a new one with
    /// [`WorkflowIdReusePolicy::Reject`](super::types::WorkflowIdReusePolicy::Reject).
    ///
    /// Raised whatever the existing workflow's status, finished included, and before anything is
    /// written. An answer rather than a failure, so it is never retried.
    WorkflowIdInUse {
        /// The id already in use.
        workflow_id: String,
        /// The existing workflow's status.
        status: WorkflowStatus,
        /// The existing workflow's name, which may be another function's.
        name: Option<String>,
    },
    /// A caller supplied a value the layer will not store.
    ///
    /// Distinct from [`Error::Malformed`], which is about values already *in* the database.
    /// This one never reaches the database at all.
    InvalidInput {
        /// The field at fault.
        field: Cow<'static, str>,
        /// What was wrong with it.
        detail: String,
    },
    /// A queued workflow already holds this deduplication key.
    ///
    /// Raised by workflow creation, and only when the caller supplied a deduplication id: the
    /// primary-key conflict is absorbed by `ON CONFLICT (workflow_uuid)`, so the sole unique
    /// violation left is the partial index on `(queue_name, deduplication_id)`. A debounced
    /// creation whose id names a workflow already holding its key raises it too, from that
    /// conflict arm. A debounce raises it as well, for a holder it must not extend.
    ///
    /// An answer rather than a failure — the caller asked to enqueue work that is already
    /// enqueued — so it is never retried.
    QueueDeduplicated {
        /// From workflow creation, the workflow that could not be enqueued. From a debounce, the
        /// workflow that holds the key: a debounce refused before it enqueues has no id of its
        /// own to name.
        workflow_id: String,
        /// The queue it was destined for.
        ///
        /// Not optional: the index is plain `UNIQUE`, so its NULLs are distinct and two rows
        /// with no queue never collide. A violation therefore proves both rows had one.
        queue_name: String,
        /// The key already held.
        deduplication_id: String,
    },
    /// The workflow has been cancelled, so its steps must not run.
    ///
    /// Raised by the step-replay check and the owner check rather than by cancellation itself:
    /// cancelling only sets a status, and a workflow already in flight learns about it the next
    /// time it asks.
    WorkflowCancelled {
        /// The cancelled workflow.
        workflow_id: String,
    },
    /// A step at this position was recorded under a different name.
    ///
    /// Means the workflow's code changed between the original run and this replay, so the
    /// recorded results no longer line up with the steps asking for them.
    UnexpectedStep {
        /// The workflow being replayed.
        workflow_id: String,
        /// The position that disagreed.
        step_id: i32,
        /// The step asking.
        expected: String,
        /// The step recorded there.
        recorded: String,
    },
    /// This execution recorded the step already, with a different result: its code isn't
    /// deterministic.
    ///
    /// No other execution can have, since the owner check runs first in the same transaction
    /// (when the caller presents its token). A retry of this caller's own write is told apart by
    /// its completion time, which a retry repeats.
    StepNondeterminism {
        /// The workflow whose step was recorded twice.
        workflow_id: String,
        /// The position that was already filled.
        step_id: i32,
    },
    /// This execution no longer owns its workflow, so it must stop and wait for the owner's
    /// outcome.
    ///
    /// The workflow was handed to another execution (resumed or recovered) and the write that found
    /// out rolled back. Every write a workflow makes from inside its execution checks first. A
    /// write refused because the workflow was cancelled is
    /// [`WorkflowCancelled`](Self::WorkflowCancelled) instead.
    WorkflowConflict {
        /// The workflow this execution no longer owns.
        workflow_id: String,
    },
    /// A workflow has no step to fork from.
    ///
    /// It exists, but nothing is recorded at the point asked for: no steps at all, or none under
    /// the name given. Forking anyway would restart it from the beginning, which is a different
    /// request from the one made.
    NoForkPoint {
        /// The workflows with nothing to fork from.
        workflow_ids: Vec<String>,
        /// The step name that matched nothing, when one was named.
        step_name: Option<String>,
    },
    /// One or more of the named workflows do not exist.
    NonExistentWorkflow {
        /// The ids with no row behind them.
        workflow_ids: Vec<String>,
    },
    /// A rewind named a workflow that is running or waiting to run.
    ///
    /// A `PENDING`, `ENQUEUED` or `DELAYED` workflow may be running or about to, and rewinding it
    /// would delete the history its own execution is replaying from. Every other status is
    /// rewindable, `MAX_RECOVERY_ATTEMPTS_EXCEEDED` included. Cancelling it first makes it
    /// rewindable.
    WorkflowNotRewindable {
        /// The workflow named.
        workflow_id: String,
        /// The status it was found in.
        status: String,
    },
    /// A rewind found the workflow terminal, but its status changed before the rewind could
    /// re-enqueue it.
    ///
    /// Something else — a resume, another rewind — moved the row while the rewind's transaction
    /// was open, and the rewind rolled back. Rewinding again sees the new status.
    RewindInterrupted {
        /// The workflow being rewound.
        workflow_id: String,
        /// The status the rewind read before the row moved.
        status: String,
    },
    /// A name is already registered by this same application.
    ///
    /// Distinct from [`Error::RegisteredByAnother`], which is a collision *between* applications
    /// and cannot be resolved here. This one the caller can resolve: update the registration, or
    /// register under a different name.
    AlreadyRegistered {
        /// What kind of thing it is, capitalised for a message: `"Schedule"`.
        kind: Cow<'static, str>,
        /// The name already taken.
        name: String,
    },
    /// A write addressed a name with no row behind it.
    NotRegistered {
        /// What kind of thing it is, capitalised for a message: `"Schedule"`.
        kind: Cow<'static, str>,
        /// The name that matched nothing.
        name: String,
    },
    /// A named thing in the system database is already registered by another application.
    ///
    /// Queue, schedule and version names address a row across every application sharing the
    /// database, so two applications cannot hold the same one. This is not the library's to
    /// resolve: taking the row would redirect a peer's work, and ignoring the write would leave
    /// this application pointing at a row it does not own.
    ///
    /// The usual causes are a genuine collision between two applications, and an application that
    /// was renamed without its rows being moved — which is what `rename_application` is for.
    RegisteredByAnother {
        /// What kind of thing it is, capitalised for a message: `"Queue"`, `"Application version"`.
        kind: Cow<'static, str>,
        /// The contested name.
        name: String,
        /// The application that holds it.
        holder: String,
        /// The application that tried to take it, if it had a name.
        claimant: Option<String>,
    },
}

impl std::fmt::Display for Error {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            Error::Backend(e) => write!(f, "system database error: {e}"),
            Error::Malformed(m) => write!(f, "unexpected value in the system database: {m}"),
            Error::UnknownSerialization { name } => {
                write!(f, "serialization {name} is not available")
            }
            Error::ConflictingWorkflow {
                workflow_id,
                detail,
            } => write!(f, "workflow {workflow_id} already exists: {detail}"),
            Error::WorkflowIdInUse {
                workflow_id,
                status,
                name,
            } => {
                write!(f, "workflow id {workflow_id} is already in use, by ")?;
                match name {
                    Some(name) => write!(f, "workflow {name}")?,
                    None => f.write_str("a workflow")?,
                }
                write!(f, " ({status})")
            }
            Error::InvalidInput { field, detail } => write!(f, "invalid {field}: {detail}"),
            Error::QueueDeduplicated {
                workflow_id,
                queue_name,
                deduplication_id,
            } => write!(
                f,
                "workflow {workflow_id} (queue: {queue_name}, \
                 deduplication id: {deduplication_id}) is already enqueued"
            ),
            Error::WorkflowCancelled { workflow_id } => {
                write!(f, "workflow {workflow_id} is cancelled")
            }
            Error::UnexpectedStep {
                workflow_id,
                step_id,
                expected,
                recorded,
            } => write!(
                f,
                "workflow {workflow_id} step {step_id} was recorded as {recorded:?}, \
                 but {expected:?} was expected"
            ),
            Error::StepNondeterminism {
                workflow_id,
                step_id,
            } => write!(
                f,
                "step {step_id} of workflow {workflow_id} was recorded twice by the same \
                 execution with different results; check that the workflow is deterministic"
            ),
            Error::WorkflowConflict { workflow_id } => {
                write!(
                    f,
                    "workflow {workflow_id} is no longer owned by this execution"
                )
            }
            Error::NoForkPoint {
                workflow_ids,
                step_name,
            } => match step_name {
                Some(name) => write!(
                    f,
                    "no step named {name} in workflows {}",
                    workflow_ids.join(", ")
                ),
                None => write!(f, "no steps in workflows {}", workflow_ids.join(", ")),
            },
            Error::NonExistentWorkflow { workflow_ids } => {
                write!(f, "no such workflow: {}", workflow_ids.join(", "))
            }
            Error::WorkflowNotRewindable {
                workflow_id,
                status,
            } => write!(
                f,
                "cannot rewind workflow {workflow_id} ({status}): it is running or waiting to \
                 run, so cancel it first"
            ),
            Error::RewindInterrupted {
                workflow_id,
                status,
            } => write!(
                f,
                "workflow {workflow_id} changed status from {status} while being rewound; \
                 retry the rewind"
            ),
            // The remedy is in the message because there is no way to act on this from code:
            // whichever cause it is, a person has to choose a name or move the rows.
            Error::AlreadyRegistered { kind, name } => {
                write!(f, "{kind} {name:?} is already registered")
            }
            Error::NotRegistered { kind, name } => {
                write!(f, "{kind} {name:?} is not registered")
            }
            Error::RegisteredByAnother {
                kind,
                name,
                holder,
                claimant,
            } => {
                let lower = kind.to_lowercase();
                write!(
                    f,
                    "{kind} {name:?} is already registered by application {holder:?} in this \
                     system database, and {lower} names must be unique across the applications \
                     sharing one"
                )?;
                if let Some(claimant) = claimant {
                    write!(
                        f,
                        ": either give {claimant:?} a different {lower} name, or, if \
                         {holder:?} was renamed to {claimant:?}, move its rows first"
                    )?;
                }
                Ok(())
            }
        }
    }
}

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

impl Error {
    /// Whether a durable step records this failure as its outcome, so a replay gives it back.
    ///
    /// Only the typed refusals are recorded — "already exists", "no such workflow", "nothing to
    /// fork from" — where asking again in the same state gets the same reply. Recording them is
    /// what keeps a replay on the branch the original run took.
    ///
    /// Never recorded:
    ///
    /// - **Any [`Backend`](Error::Backend) failure.** Connection loss and conflicts are accidents
    ///   of the moment: the work rolled back, and [`with_retry`](super::retry::with_retry) runs it
    ///   again. A permanent one is the database or the build rather than an answer — a schema
    ///   behind the build, a missing grant, a column the driver cannot decode — because every
    ///   answer the schema can give is already mapped to a typed variant. Each can be fixed after
    ///   the fact, and a recorded failure would outlive the fix.
    /// - **A value this build cannot read.** That says something about the build rather than the
    ///   data; a later build may read it, and a recorded failure would outlive the fix.
    /// - **Signals about the calling execution** — cancelled, no longer the workflow's owner, or
    ///   replaying code that has changed — which end the run rather than answer the call.
    ///
    /// Exhaustive, so a new variant has to be placed on one side or the other.
    pub(crate) fn should_record(&self) -> bool {
        match self {
            Error::Backend(_)
            | Error::Malformed(_)
            | Error::UnknownSerialization { .. }
            | Error::WorkflowCancelled { .. }
            | Error::StepNondeterminism { .. }
            | Error::WorkflowConflict { .. }
            | Error::UnexpectedStep { .. } => false,
            Error::ConflictingWorkflow { .. }
            | Error::WorkflowIdInUse { .. }
            | Error::InvalidInput { .. }
            | Error::QueueDeduplicated { .. }
            | Error::NoForkPoint { .. }
            | Error::NonExistentWorkflow { .. }
            | Error::WorkflowNotRewindable { .. }
            | Error::RewindInterrupted { .. }
            | Error::AlreadyRegistered { .. }
            | Error::NotRegistered { .. }
            | Error::RegisteredByAnother { .. } => true,
        }
    }
}

/// A failure the database or its driver reported.
#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub struct BackendError {
    /// What the driver said.
    pub message: String,
    /// The SQLSTATE, when the failure came from the database rather than the connection.
    pub sqlstate: Option<String>,
    /// Whether waiting and asking again could succeed.
    pub kind: BackendErrorKind,
}

impl std::fmt::Display for BackendError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match &self.sqlstate {
            Some(code) => write!(f, "{} ({code})", self.message),
            None => f.write_str(&self.message),
        }
    }
}

/// Whether a backend failure is worth asking again about.
///
/// Classification is the backend's job, not the retry loop's: SQLSTATEs are Postgres's, and a
/// SQLite backend would decide on message text instead.
#[derive(Debug, Clone, Copy, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub enum BackendErrorKind {
    /// The connection failed, or the server cannot serve requests right now.
    ///
    /// Retried by default. A caller that would rather see the failure than block can opt out
    /// with [`super::retry::RetryPolicy::retry_connection_errors`].
    Connection,
    /// A transaction conflict the database has already rolled back: a serialization failure or a
    /// deadlock.
    ///
    /// The database is working; it asked this transaction to step aside so another could commit.
    /// Replaying it is safe because nothing of it was kept. Replayed by default; a caller can opt
    /// out with [`super::retry::RetryPolicy::retry_conflicts`].
    Conflict,
    /// The database understood the request and rejected it.
    ///
    /// A syntax error, a constraint violation, a missing table. Asking again cannot help.
    Permanent,
}

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

    fn backend(kind: BackendErrorKind) -> Error {
        Error::Backend(BackendError {
            message: "boom".to_owned(),
            sqlstate: None,
            kind,
        })
    }

    #[test]
    fn a_typed_refusal_should_be_recorded() {
        assert!(
            Error::NonExistentWorkflow {
                workflow_ids: vec!["wf".to_owned()],
            }
            .should_record()
        );
        assert!(
            Error::AlreadyRegistered {
                kind: "Schedule".into(),
                name: "nightly".to_owned(),
            }
            .should_record()
        );
        // Both of a rewind's refusals. The interrupted one is recorded too: unrecorded, a replay
        // would run the rewind again, and could rewind a workflow the original run did not.
        assert!(
            Error::WorkflowNotRewindable {
                workflow_id: "wf".to_owned(),
                status: "PENDING".to_owned(),
            }
            .should_record()
        );
        assert!(
            Error::RewindInterrupted {
                workflow_id: "wf".to_owned(),
                status: "SUCCESS".to_owned(),
            }
            .should_record()
        );
    }

    #[test]
    fn a_backend_failure_or_a_signal_should_not_be_recorded() {
        assert!(!backend(BackendErrorKind::Permanent).should_record());
        assert!(!backend(BackendErrorKind::Connection).should_record());
        assert!(!backend(BackendErrorKind::Conflict).should_record());
        assert!(!Error::Malformed("unreadable".to_owned()).should_record());
        assert!(
            !Error::UnknownSerialization {
                name: "pickle".to_owned(),
            }
            .should_record()
        );
        assert!(
            !Error::WorkflowCancelled {
                workflow_id: "wf".to_owned(),
            }
            .should_record()
        );
        assert!(
            !Error::StepNondeterminism {
                workflow_id: "wf".to_owned(),
                step_id: 0,
            }
            .should_record()
        );
        assert!(
            !Error::WorkflowConflict {
                workflow_id: "wf".to_owned(),
            }
            .should_record()
        );
    }
}