greentic-operator 0.4.43

Greentic operator CLI for local dev and demo orchestration.
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
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120

//! Retry scheduling helpers for the universal pipeline.

use chrono::Utc;
use greentic_types::ChannelMessageEnvelope;
use serde::{Deserialize, Serialize};
use serde_json::Value;
use std::time::Duration;
use uuid::Uuid;

#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct RetryPolicy {
    pub max_attempts: u32,
    pub base_delay_ms: u64,
    pub max_delay_ms: u64,
    pub jitter_ms: u64,
}

impl Default for RetryPolicy {
    fn default() -> Self {
        Self {
            max_attempts: 5,
            base_delay_ms: 500,
            max_delay_ms: 30_000,
            jitter_ms: 250,
        }
    }
}

impl RetryPolicy {
    pub fn backoff_ms(&self, attempt: u32) -> u64 {
        let shift = attempt.saturating_sub(1).min(63);
        let multiplier = 1u64 << shift;
        let base = self.base_delay_ms.saturating_mul(multiplier);
        std::cmp::min(base, self.max_delay_ms)
    }

    pub fn delay_with_jitter(&self, attempt: u32, jitter_ms: u64) -> Duration {
        let backoff = self.backoff_ms(attempt);
        let jitter = std::cmp::min(jitter_ms, self.jitter_ms);
        Duration::from_millis(backoff.saturating_add(jitter))
    }
}

#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct EgressJob {
    pub job_id: Uuid,
    pub provider: String,
    pub attempt: u32,
    pub max_attempts: u32,
    pub next_run_at_unix_ms: u64,
    pub envelope: ChannelMessageEnvelope,
    pub plan_cache: Option<Value>,
    pub last_error: Option<String>,
}

impl EgressJob {
    pub fn new(
        provider: impl Into<String>,
        envelope: ChannelMessageEnvelope,
        max_attempts: u32,
    ) -> Self {
        Self {
            job_id: Uuid::new_v4(),
            provider: provider.into(),
            attempt: 0,
            max_attempts: max_attempts.max(1),
            next_run_at_unix_ms: current_time_ms(),
            envelope,
            plan_cache: None,
            last_error: None,
        }
    }

    pub fn increment_attempt(&mut self) {
        self.attempt = self.attempt.saturating_add(1);
    }

    pub fn schedule_next(&mut self, delay_ms: u64) {
        self.next_run_at_unix_ms = current_time_ms().saturating_add(delay_ms);
    }

    pub fn with_plan(&mut self, plan: serde_json::Value) {
        self.plan_cache = Some(plan);
    }

    pub fn record_error(&mut self, error: String) {
        self.last_error = Some(error);
    }
}

fn current_time_ms() -> u64 {
    Utc::now().timestamp_millis().max(0) as u64
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn backoff_steps_up_to_max() {
        let policy = RetryPolicy {
            base_delay_ms: 100,
            max_delay_ms: 400,
            ..Default::default()
        };
        assert_eq!(policy.backoff_ms(1), 100);
        assert_eq!(policy.backoff_ms(2), 200);
        assert_eq!(policy.backoff_ms(3), 400);
        assert_eq!(policy.backoff_ms(4), 400);
    }

    #[test]
    fn delay_injects_jitter() {
        let policy = RetryPolicy::default();
        let without_jitter = policy.delay_with_jitter(1, 0);
        let with_jitter = policy.delay_with_jitter(1, 123);
        assert!(with_jitter >= without_jitter);
        assert!(with_jitter <= without_jitter + Duration::from_millis(policy.jitter_ms));
    }
}