obeli_sk_executor/

worker.rs

use async_trait::async_trait;
use concepts::ExecutionFailureKind;
use concepts::ExecutionId;
use concepts::ExecutionMetadata;
use concepts::FunctionMetadata;
use concepts::JoinSetId;
use concepts::TrapKind;
use concepts::component_id::ComponentDigest;
use concepts::storage::DbErrorWrite;
use concepts::storage::HistoryEvent;
use concepts::storage::Locked;
use concepts::storage::ResponseWithCursor;
use concepts::storage::Version;
use concepts::storage::http_client_trace::HttpClientTrace;
use concepts::{FinishedExecutionFailure, StrVariant};
use concepts::{FunctionFqn, ParamsParsingError, ResultParsingError};
use concepts::{Params, SupportedFunctionReturnValue};
use tracing::Span;

#[async_trait]
pub trait Worker: Send + Sync + 'static {
    async fn run(&self, ctx: WorkerContext) -> WorkerResult;

    // List exported functions without extensions.
    // Used by executor.
    fn exported_functions_noext(&self) -> &[FunctionMetadata];
}

pub type WorkerResult = Result<WorkerResultOk, WorkerError>;

#[derive(Debug, derive_more::Display)]
pub enum WorkerResultOk {
    #[display("db updated, not finished")]
    DbUpdatedByWorkerOrWatcher,
    /// The execution run has returned a valid `retval`. Activity with retry budget returning `err` will be retried by the executor.
    #[display("{_0}")]
    RunFinished(RunFinished),
}

#[derive(Debug, Clone, Copy, derive_more::Display)]
pub enum ExecutionYieldReason {
    #[display("executor closing")]
    ExecutorClosing,
    #[display("workflow event limit reached")]
    WorkflowEventLimitReached,
}

#[derive(Debug, derive_more::Display)]
#[display("finished at v{version}")]
pub struct RunFinished {
    pub retval: SupportedFunctionReturnValue,
    pub version: Version,
    pub http_client_traces: Option<Vec<HttpClientTrace>>,
}

#[derive(Debug)]
pub struct WorkerContext {
    pub execution_id: ExecutionId,
    pub metadata: ExecutionMetadata,
    pub component_digest: ComponentDigest,
    pub ffqn: FunctionFqn,
    pub params: Params,
    pub event_history: Vec<(HistoryEvent, Version)>,
    pub responses: Vec<ResponseWithCursor>,
    pub parent: Option<(ExecutionId, JoinSetId)>,
    pub version: Version,
    pub can_be_retried: bool,
    pub worker_span: Span,
    pub locked_event: Locked,
    pub execution_interrupt_watcher: tokio::sync::watch::Receiver<bool>,
}

#[derive(Debug, thiserror::Error)]
pub enum WorkerError {
    // retriable errors
    // Used by activity worker
    #[error("activity {trap_kind}: {reason}")]
    ActivityTrap {
        reason: String,
        trap_kind: TrapKind,
        detail: Option<String>,
        version: Version,
        http_client_traces: Option<Vec<HttpClientTrace>>,
    },
    /// Resources are exhausted.
    /// Executor must mark the execution as Unlocked.
    /// This event does not increase temporary event count.
    #[error("limit reached: {reason}")]
    LimitReached { reason: String, version: Version },
    // Used by activity or workflow worker, best effort. If this is not persisted, the expired timers watcher will append it.
    #[error("temporary timeout")]
    TemporaryTimeout {
        http_client_traces: Option<Vec<HttpClientTrace>>,
        version: Version,
    },
    #[error("execution yielded: {reason}")]
    ExecutionYielded {
        version: Version,
        reason: ExecutionYieldReason,
    },
    #[error(transparent)]
    DbError(DbErrorWrite),
    // non-retriable errors
    #[error("fatal error: {0}")]
    FatalError(FatalError, Version),
}

#[derive(Debug, thiserror::Error)]
pub enum FatalError {
    // Used by workflow worker
    #[error("nondeterminism detected")]
    NondeterminismDetected { detail: String },
    // Used by activity worker, workflow worker
    #[error(transparent)]
    ParamsParsingError(ParamsParsingError),
    // Used by activity worker, workflow worker
    #[error("{reason}")]
    CannotInstantiate {
        reason: String,
        detail: Option<String>,
    },
    // Used by activity worker, workflow worker
    #[error(transparent)]
    ResultParsingError(ResultParsingError),
    /// Used when workflow cannot call an imported function, either a child execution or a function from workflow-support.
    #[error("error calling imported function {ffqn} : {reason}")]
    ImportedFunctionCallError {
        ffqn: FunctionFqn,
        reason: StrVariant,
        detail: Option<String>,
    },
    /// Workflow trap if `retry_on_trap` is disabled.
    #[error("workflow {trap_kind}: {reason}")]
    WorkflowTrap {
        reason: String,
        trap_kind: TrapKind,
        detail: Option<String>,
    },
    #[error("out of fuel: {reason}")]
    OutOfFuel { reason: String },
    #[error("constraint violation: {reason}")]
    ConstraintViolation { reason: StrVariant },
    // Applies to activities.
    #[error("cancelled")]
    Cancelled,
}

impl From<FatalError> for FinishedExecutionFailure {
    fn from(err: FatalError) -> Self {
        let reason_generic = err.to_string(); // Override with err's reason if no information is lost.
        match err {
            FatalError::NondeterminismDetected { detail } => FinishedExecutionFailure {
                reason: None,
                kind: ExecutionFailureKind::NondeterminismDetected,
                detail: Some(detail),
            },
            FatalError::OutOfFuel { reason } => FinishedExecutionFailure {
                reason: Some(reason),
                kind: ExecutionFailureKind::OutOfFuel,
                detail: None,
            },
            FatalError::ParamsParsingError(err) => FinishedExecutionFailure {
                reason: Some(reason_generic),
                kind: ExecutionFailureKind::Uncategorized,
                detail: err.detail(),
            },
            FatalError::CannotInstantiate { reason, detail } => FinishedExecutionFailure {
                reason: Some(reason),
                kind: ExecutionFailureKind::Uncategorized,
                detail,
            },
            FatalError::ResultParsingError(_) | FatalError::ConstraintViolation { reason: _ } => {
                FinishedExecutionFailure {
                    reason: Some(reason_generic),
                    kind: ExecutionFailureKind::Uncategorized,
                    detail: None,
                }
            }
            FatalError::ImportedFunctionCallError { detail, .. }
            | FatalError::WorkflowTrap { detail, .. } => FinishedExecutionFailure {
                reason: Some(reason_generic),
                kind: ExecutionFailureKind::Uncategorized,
                detail,
            },
            FatalError::Cancelled => FinishedExecutionFailure {
                kind: ExecutionFailureKind::Cancelled,
                reason: None,
                detail: None,
            },
        }
    }
}

impl From<&FatalError> for FinishedExecutionFailure {
    fn from(err: &FatalError) -> Self {
        let reason_generic = err.to_string(); // Override with err's reason if no information is lost.
        match err {
            FatalError::NondeterminismDetected { detail } => FinishedExecutionFailure {
                reason: None,
                kind: ExecutionFailureKind::NondeterminismDetected,
                detail: Some(detail.clone()),
            },
            FatalError::OutOfFuel { reason } => FinishedExecutionFailure {
                reason: Some(reason.clone()),
                kind: ExecutionFailureKind::OutOfFuel,
                detail: None,
            },
            FatalError::ParamsParsingError(err) => FinishedExecutionFailure {
                reason: Some(reason_generic),
                kind: ExecutionFailureKind::Uncategorized,
                detail: err.detail(),
            },
            FatalError::CannotInstantiate { reason, detail } => FinishedExecutionFailure {
                reason: Some(reason.clone()),
                kind: ExecutionFailureKind::Uncategorized,
                detail: detail.clone(),
            },
            FatalError::ResultParsingError(_) | FatalError::ConstraintViolation { reason: _ } => {
                FinishedExecutionFailure {
                    reason: Some(reason_generic),
                    kind: ExecutionFailureKind::Uncategorized,
                    detail: None,
                }
            }
            FatalError::ImportedFunctionCallError { detail, .. }
            | FatalError::WorkflowTrap { detail, .. } => FinishedExecutionFailure {
                reason: Some(reason_generic),
                kind: ExecutionFailureKind::Uncategorized,
                detail: detail.clone(),
            },
            FatalError::Cancelled => FinishedExecutionFailure {
                kind: ExecutionFailureKind::Cancelled,
                reason: None,
                detail: None,
            },
        }
    }
}