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)]
30pub enum WorkerResultOk {
31    DbUpdatedByWorkerOrWatcher,
32    Finished {
33        retval: SupportedFunctionReturnValue,
34        version: Version,
35        http_client_traces: Option<Vec<HttpClientTrace>>,
36    },
37}
38
39#[derive(Debug)]
40pub struct WorkerContext {
41    pub execution_id: ExecutionId,
42    pub metadata: ExecutionMetadata,
43    pub ffqn: FunctionFqn,
44    pub params: Params,
45    pub event_history: Vec<(HistoryEvent, Version)>,
46    pub responses: Vec<ResponseWithCursor>,
47    pub version: Version,
48    pub can_be_retried: bool,
49    pub worker_span: Span,
50    pub locked_event: Locked,
51}
52
53#[derive(Debug, thiserror::Error)]
54pub enum WorkerError {
55    // retriable errors
56    // Used by activity worker
57    #[error("activity {trap_kind}: {reason}")]
58    ActivityTrap {
59        reason: String,
60        trap_kind: TrapKind,
61        detail: Option<String>,
62        version: Version,
63        http_client_traces: Option<Vec<HttpClientTrace>>,
64    },
65    #[error("{reason}")]
66    ActivityPreopenedDirError {
67        reason: String,
68        detail: String,
69        version: Version,
70    },
71    // Used by activity worker, must not be returned when retries are exhausted.
72    #[error("activity returned error")]
73    ActivityReturnedError {
74        detail: Option<String>,
75        version: Version,
76        http_client_traces: Option<Vec<HttpClientTrace>>,
77    },
78    /// Resources are exhausted.
79    /// Executor must mark the execution as Unlocked.
80    /// This event does not increase temporary event count.
81    #[error("limit reached: {reason}")]
82    LimitReached { reason: String, version: Version },
83    // Used by activity worker, best effort. If this is not persisted, the expired timers watcher will append it.
84    #[error("temporary timeout")]
85    TemporaryTimeout {
86        http_client_traces: Option<Vec<HttpClientTrace>>,
87        version: Version,
88    },
89    #[error(transparent)]
90    DbError(DbErrorWrite),
91    // non-retriable errors
92    #[error("fatal error: {0}")]
93    FatalError(FatalError, Version),
94}
95
96#[derive(Debug, thiserror::Error)]
97pub enum FatalError {
98    // Used by workflow worker
99    #[error("nondeterminism detected")]
100    NondeterminismDetected { detail: String },
101    // Used by activity worker, workflow worker
102    #[error(transparent)]
103    ParamsParsingError(ParamsParsingError),
104    // Used by activity worker, workflow worker
105    #[error("{reason}")]
106    CannotInstantiate {
107        reason: String,
108        detail: Option<String>,
109    },
110    // Used by activity worker, workflow worker
111    #[error(transparent)]
112    ResultParsingError(ResultParsingError),
113    /// Used when workflow cannot call an imported function, either a child execution or a function from workflow-support.
114    #[error("error calling imported function {ffqn} : {reason}")]
115    ImportedFunctionCallError {
116        ffqn: FunctionFqn,
117        reason: StrVariant,
118        detail: Option<String>,
119    },
120    /// Workflow trap if `retry_on_trap` is disabled.
121    #[error("workflow {trap_kind}: {reason}")]
122    WorkflowTrap {
123        reason: String,
124        trap_kind: TrapKind,
125        detail: Option<String>,
126    },
127    #[error("out of fuel: {reason}")]
128    OutOfFuel { reason: String },
129    #[error("constraint violation: {reason}")]
130    ConstraintViolation { reason: StrVariant },
131    // Applies to activities.
132    #[error("cancelled")]
133    Cancelled,
134}
135
136impl From<FatalError> for FinishedExecutionError {
137    fn from(err: FatalError) -> Self {
138        let reason_generic = err.to_string(); // Override with err's reason if no information is lost.
139        match err {
140            FatalError::NondeterminismDetected { detail } => FinishedExecutionError {
141                reason: None,
142                kind: ExecutionFailureKind::NondeterminismDetected,
143                detail: Some(detail),
144            },
145            FatalError::OutOfFuel { reason } => FinishedExecutionError {
146                reason: Some(reason),
147                kind: ExecutionFailureKind::OutOfFuel,
148                detail: None,
149            },
150            FatalError::ParamsParsingError(err) => FinishedExecutionError {
151                reason: Some(reason_generic),
152                kind: ExecutionFailureKind::Uncategorized,
153                detail: err.detail(),
154            },
155            FatalError::CannotInstantiate { reason, detail } => FinishedExecutionError {
156                reason: Some(reason),
157                kind: ExecutionFailureKind::Uncategorized,
158                detail,
159            },
160            FatalError::ResultParsingError(_) | FatalError::ConstraintViolation { reason: _ } => {
161                FinishedExecutionError {
162                    reason: Some(reason_generic),
163                    kind: ExecutionFailureKind::Uncategorized,
164                    detail: None,
165                }
166            }
167            FatalError::ImportedFunctionCallError { detail, .. }
168            | FatalError::WorkflowTrap { detail, .. } => FinishedExecutionError {
169                reason: Some(reason_generic),
170                kind: ExecutionFailureKind::Uncategorized,
171                detail,
172            },
173            FatalError::Cancelled => FinishedExecutionError {
174                kind: ExecutionFailureKind::Cancelled,
175                reason: None,
176                detail: None,
177            },
178        }
179    }
180}
obeli_sk_executor/worker.rs

obeli_sk_executor/
worker.rs
