obeli_sk_executor/
worker.rs

1use async_trait::async_trait;
2use concepts::ExecutionFailureKind;
3use concepts::ExecutionId;
4use concepts::ExecutionMetadata;
5use concepts::FunctionMetadata;
6use concepts::TrapKind;
7use concepts::storage::DbErrorWrite;
8use concepts::storage::HistoryEvent;
9use concepts::storage::Locked;
10use concepts::storage::ResponseWithCursor;
11use concepts::storage::Version;
12use concepts::storage::http_client_trace::HttpClientTrace;
13use concepts::{FinishedExecutionError, StrVariant};
14use concepts::{FunctionFqn, ParamsParsingError, ResultParsingError};
15use concepts::{Params, SupportedFunctionReturnValue};
16use tracing::Span;
17
18#[async_trait]
19pub trait Worker: Send + Sync + 'static {
20    async fn run(&self, ctx: WorkerContext) -> WorkerResult;
21
22    // List exported functions without extensions.
23    // Used by executor.
24    fn exported_functions_noext(&self) -> &[FunctionMetadata];
25}
26
27pub type WorkerResult = Result<WorkerResultOk, WorkerError>;
28
29#[derive(Debug, derive_more::Display)]
30pub enum WorkerResultOk {
31    #[display("db updated by worker or watcher")]
32    DbUpdatedByWorkerOrWatcher,
33    /// The execution run has returned a valid `retval`. Activity with retry budget returning `err` will be retried by the executor.
34    #[display("{retval}")]
35    RunFinished {
36        retval: SupportedFunctionReturnValue,
37        version: Version,
38        http_client_traces: Option<Vec<HttpClientTrace>>,
39    },
40}
41
42#[derive(Debug)]
43pub struct WorkerContext {
44    pub execution_id: ExecutionId,
45    pub metadata: ExecutionMetadata,
46    pub ffqn: FunctionFqn,
47    pub params: Params,
48    pub event_history: Vec<(HistoryEvent, Version)>,
49    pub responses: Vec<ResponseWithCursor>,
50    pub version: Version,
51    pub can_be_retried: bool,
52    pub worker_span: Span,
53    pub locked_event: Locked,
54    pub executor_close_watcher: tokio::sync::watch::Receiver<bool>,
55}
56
57#[derive(Debug, thiserror::Error)]
58pub enum WorkerError {
59    // retriable errors
60    // Used by activity worker
61    #[error("activity {trap_kind}: {reason}")]
62    ActivityTrap {
63        reason: String,
64        trap_kind: TrapKind,
65        detail: Option<String>,
66        version: Version,
67        http_client_traces: Option<Vec<HttpClientTrace>>,
68    },
69    #[error("{reason}")]
70    ActivityPreopenedDirError {
71        reason: String,
72        detail: String,
73        version: Version,
74    },
75    /// Resources are exhausted.
76    /// Executor must mark the execution as Unlocked.
77    /// This event does not increase temporary event count.
78    #[error("limit reached: {reason}")]
79    LimitReached { reason: String, version: Version },
80    // Used by activity or workflow worker, best effort. If this is not persisted, the expired timers watcher will append it.
81    #[error("temporary timeout")]
82    TemporaryTimeout {
83        http_client_traces: Option<Vec<HttpClientTrace>>,
84        version: Version,
85    },
86    #[error("executor closing")]
87    ExecutorClosing(Version),
88    #[error(transparent)]
89    DbError(DbErrorWrite),
90    // non-retriable errors
91    #[error("fatal error: {0}")]
92    FatalError(FatalError, Version),
93}
94
95#[derive(Debug, thiserror::Error)]
96pub enum FatalError {
97    // Used by workflow worker
98    #[error("nondeterminism detected")]
99    NondeterminismDetected { detail: String },
100    // Used by activity worker, workflow worker
101    #[error(transparent)]
102    ParamsParsingError(ParamsParsingError),
103    // Used by activity worker, workflow worker
104    #[error("{reason}")]
105    CannotInstantiate {
106        reason: String,
107        detail: Option<String>,
108    },
109    // Used by activity worker, workflow worker
110    #[error(transparent)]
111    ResultParsingError(ResultParsingError),
112    /// Used when workflow cannot call an imported function, either a child execution or a function from workflow-support.
113    #[error("error calling imported function {ffqn} : {reason}")]
114    ImportedFunctionCallError {
115        ffqn: FunctionFqn,
116        reason: StrVariant,
117        detail: Option<String>,
118    },
119    /// Workflow trap if `retry_on_trap` is disabled.
120    #[error("workflow {trap_kind}: {reason}")]
121    WorkflowTrap {
122        reason: String,
123        trap_kind: TrapKind,
124        detail: Option<String>,
125    },
126    #[error("out of fuel: {reason}")]
127    OutOfFuel { reason: String },
128    #[error("constraint violation: {reason}")]
129    ConstraintViolation { reason: StrVariant },
130    // Applies to activities.
131    #[error("cancelled")]
132    Cancelled,
133}
134
135impl From<FatalError> for FinishedExecutionError {
136    fn from(err: FatalError) -> Self {
137        let reason_generic = err.to_string(); // Override with err's reason if no information is lost.
138        match err {
139            FatalError::NondeterminismDetected { detail } => FinishedExecutionError {
140                reason: None,
141                kind: ExecutionFailureKind::NondeterminismDetected,
142                detail: Some(detail),
143            },
144            FatalError::OutOfFuel { reason } => FinishedExecutionError {
145                reason: Some(reason),
146                kind: ExecutionFailureKind::OutOfFuel,
147                detail: None,
148            },
149            FatalError::ParamsParsingError(err) => FinishedExecutionError {
150                reason: Some(reason_generic),
151                kind: ExecutionFailureKind::Uncategorized,
152                detail: err.detail(),
153            },
154            FatalError::CannotInstantiate { reason, detail } => FinishedExecutionError {
155                reason: Some(reason),
156                kind: ExecutionFailureKind::Uncategorized,
157                detail,
158            },
159            FatalError::ResultParsingError(_) | FatalError::ConstraintViolation { reason: _ } => {
160                FinishedExecutionError {
161                    reason: Some(reason_generic),
162                    kind: ExecutionFailureKind::Uncategorized,
163                    detail: None,
164                }
165            }
166            FatalError::ImportedFunctionCallError { detail, .. }
167            | FatalError::WorkflowTrap { detail, .. } => FinishedExecutionError {
168                reason: Some(reason_generic),
169                kind: ExecutionFailureKind::Uncategorized,
170                detail,
171            },
172            FatalError::Cancelled => FinishedExecutionError {
173                kind: ExecutionFailureKind::Cancelled,
174                reason: None,
175                detail: None,
176            },
177        }
178    }
179}
obeli_sk_executor/worker.rs

obeli_sk_executor/
worker.rs
