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

obeli_sk_executor/
worker.rs
