1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75

use async_trait::async_trait;
use chrono::{DateTime, Utc};
use concepts::storage::HistoryEvent;
use concepts::storage::Version;
use concepts::ExecutionId;
use concepts::FunctionMetadata;
use concepts::{
    storage::{DbError, JoinSetResponseEvent},
    FinishedExecutionError, StrVariant,
};
use concepts::{FunctionFqn, ParamsParsingError, ResultParsingError};
use concepts::{Params, SupportedFunctionResult};
use std::error::Error;

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

    fn exported_functions(&self) -> impl Iterator<Item = FunctionMetadata>;

    fn imported_functions(&self) -> impl Iterator<Item = FunctionMetadata>;
}

#[must_use]
#[derive(Debug)]
pub enum WorkerResult {
    Ok(SupportedFunctionResult, Version),
    ChildExecutionRequest,
    DelayRequest,
    Err(WorkerError),
}

#[derive(Debug)]
pub struct WorkerContext {
    pub execution_id: ExecutionId,
    pub ffqn: FunctionFqn,
    pub params: Params,
    pub event_history: Vec<HistoryEvent>,
    pub responses: Vec<JoinSetResponseEvent>,
    pub version: Version,
    pub execution_deadline: DateTime<Utc>,
    pub can_be_retried: bool,
}

#[derive(Debug, thiserror::Error)]
pub enum WorkerError {
    #[error("intermittent error: {reason}")]
    IntermittentError {
        reason: StrVariant,
        err: Option<Box<dyn Error + Send + Sync>>,
        version: Version,
    },
    #[error("Limit reached: {0}")]
    LimitReached(String, Version),
    #[error("intermittent timeout")]
    IntermittentTimeout,
    #[error("fatal error: {0}")]
    FatalError(FatalError, Version),
    #[error(transparent)]
    DbError(DbError),
}

#[derive(Debug, thiserror::Error)]
pub enum FatalError {
    #[error("non-determinism detected: `{0}`")]
    NonDeterminismDetected(StrVariant),
    #[error("parameters cannot be parsed: {0}")]
    ParamsParsingError(ParamsParsingError),
    #[error("result cannot be parsed: {0}")]
    ResultParsingError(ResultParsingError),
    #[error("child finished with an execution error: {0}")]
    ChildExecutionError(FinishedExecutionError),
    #[error("uncategorized error: {0}")]
    UncategorizedError(&'static str),
}