jamjet-core 0.3.2

JamJet core execution model — workflow states, node types, retry/timeout policies
Documentation 
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
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92

use serde::{Deserialize, Deserializer, Serialize, Serializer};
use std::time::Duration;

/// Timeout configuration for a workflow execution.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct TimeoutConfig {
    /// Maximum time a single node may run before being killed and failed.
    #[serde(
        default,
        serialize_with = "ser_opt_duration",
        deserialize_with = "de_opt_duration"
    )]
    pub node_timeout: Option<Duration>,
    /// Maximum total time a workflow execution may run.
    #[serde(
        default,
        serialize_with = "ser_opt_duration",
        deserialize_with = "de_opt_duration"
    )]
    pub workflow_timeout: Option<Duration>,
    /// How often a worker must renew its lease heartbeat.
    /// If a worker misses this, the lease is reclaimed and the node is re-queued.
    #[serde(
        default = "default_heartbeat",
        serialize_with = "ser_duration",
        deserialize_with = "de_duration"
    )]
    pub heartbeat_interval: Duration,
    /// Maximum time a human_approval node waits before routing to fallback or failing.
    #[serde(
        default,
        serialize_with = "ser_opt_duration",
        deserialize_with = "de_opt_duration"
    )]
    pub approval_timeout: Option<Duration>,
}

fn default_heartbeat() -> Duration {
    Duration::from_secs(30)
}

impl Default for TimeoutConfig {
    fn default() -> Self {
        Self {
            node_timeout: Some(Duration::from_secs(300)), // 5 min
            workflow_timeout: None,
            heartbeat_interval: default_heartbeat(),
            approval_timeout: None,
        }
    }
}

// Serialize Duration as seconds (integer).
fn ser_duration<S: Serializer>(d: &Duration, s: S) -> Result<S::Ok, S::Error> {
    s.serialize_u64(d.as_secs())
}

fn ser_opt_duration<S: Serializer>(d: &Option<Duration>, s: S) -> Result<S::Ok, S::Error> {
    match d {
        Some(d) => s.serialize_some(&d.as_secs()),
        None => s.serialize_none(),
    }
}

// Deserialize Duration from either integer (seconds) or Duration struct.
fn de_duration<'de, D: Deserializer<'de>>(d: D) -> Result<Duration, D::Error> {
    #[derive(Deserialize)]
    #[serde(untagged)]
    enum DurationRepr {
        Secs(u64),
        Struct { secs: u64, nanos: u32 },
    }
    match DurationRepr::deserialize(d)? {
        DurationRepr::Secs(s) => Ok(Duration::from_secs(s)),
        DurationRepr::Struct { secs, nanos } => Ok(Duration::new(secs, nanos)),
    }
}

fn de_opt_duration<'de, D: Deserializer<'de>>(d: D) -> Result<Option<Duration>, D::Error> {
    #[derive(Deserialize)]
    #[serde(untagged)]
    enum OptDurationRepr {
        Null,
        Secs(u64),
        Struct { secs: u64, nanos: u32 },
    }
    match Option::<OptDurationRepr>::deserialize(d)? {
        None | Some(OptDurationRepr::Null) => Ok(None),
        Some(OptDurationRepr::Secs(s)) => Ok(Some(Duration::from_secs(s))),
        Some(OptDurationRepr::Struct { secs, nanos }) => Ok(Some(Duration::new(secs, nanos))),
    }
}