aion-rs 0.10.0

Transport-agnostic Aion workflow engine with durability, replay, timers, and supervision.
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

use super::{await_support::envelope, support::*};

// ---- #197: retry loop at the dispatch seam ------------------------------

/// Scripted dispatcher for the retry-loop tests: pops one outcome per
/// dispatch and records the wire attempt each delivery carried.
pub(in super::super) struct ScriptedRetryDispatcher {
    outcomes: std::sync::Mutex<std::collections::VecDeque<Result<String, String>>>,
    attempts: std::sync::Mutex<Vec<u32>>,
}

impl ScriptedRetryDispatcher {
    pub(super) fn new(outcomes: Vec<Result<String, String>>) -> Arc<Self> {
        Arc::new(Self {
            outcomes: std::sync::Mutex::new(outcomes.into_iter().collect()),
            attempts: std::sync::Mutex::new(Vec::new()),
        })
    }

    pub(super) fn seen_attempts(&self) -> Vec<u32> {
        self.attempts
            .lock()
            .map(|attempts| attempts.clone())
            .unwrap_or_default()
    }
}

impl ActivityDispatcher for ScriptedRetryDispatcher {
    fn dispatch(&self, request: ActivityDispatch) -> Result<String, String> {
        self.attempts
            .lock()
            .map_err(|_| "attempts lock poisoned".to_owned())?
            .push(request.attempt);
        self.outcomes
            .lock()
            .map_err(|_| "outcomes lock poisoned".to_owned())?
            .pop_front()
            .ok_or_else(|| "terminal:script exhausted — unexpected extra dispatch".to_owned())?
    }
}

/// Store + recorder + request over a seeded `WorkflowStarted` +
/// `ActivityScheduled` + `ActivityStarted(attempt 1)` history — exactly
/// what the dispatch NIF records before spawning the completion task.
pub(in super::super) struct RetryLoopHarness {
    pub(super) store: Arc<dyn EventStore>,
    pub(super) seam: RetryRecorderSeam,
    pub(super) request: ActivityDispatch,
    pub(super) workflow_id: WorkflowId,
}

impl RetryLoopHarness {
    pub(super) async fn seeded(config: &str) -> Result<Self, Box<dyn std::error::Error>> {
        let store: Arc<dyn EventStore> = Arc::new(aion_store::InMemoryStore::default());
        let workflow_id = WorkflowId::new_v4();
        let run_id = RunId::new_v4();
        let events = vec![
            Event::WorkflowStarted {
                envelope: envelope(&workflow_id, 1),
                workflow_type: "retrier".to_owned(),
                input: Payload::from_json(&json!({}))?,
                run_id: run_id.clone(),
                parent_run_id: None,
                package_version: aion_core::PackageVersion::new("b".repeat(64)),
            },
            Event::ActivityScheduled {
                envelope: envelope(&workflow_id, 2),
                activity_id: ActivityId::from_sequence_position(0),
                activity_type: "flaky".to_owned(),
                input: Payload::new(ContentType::Json, br#""in""#.to_vec()),
                task_queue: String::from("default"),
                node: None,
            },
            Event::ActivityStarted {
                envelope: envelope(&workflow_id, 3),
                activity_id: ActivityId::from_sequence_position(0),
                attempt: 1,
            },
        ];
        store
            .append(WriteToken::recorder(), &workflow_id, &events, 0)
            .await?;
        let recorder = Recorder::resume_at(workflow_id.clone(), Arc::clone(&store), 3);
        let request = ActivityDispatch {
            namespace: String::from("default"),
            task_queue: String::from("default"),
            node: None,
            workflow_id: workflow_id.clone(),
            activity_id: ActivityId::from_sequence_position(0),
            name: "flaky".to_owned(),
            input: r#""in""#.to_owned(),
            config: config.to_owned(),
            attempt: super::FIRST_DELIVERY_ATTEMPT,
            labels: std::collections::BTreeMap::new(),
        };
        Ok(Self {
            store,
            seam: RetryRecorderSeam {
                recorder: Arc::new(tokio::sync::Mutex::new(recorder)),
                run_id,
            },
            request,
            workflow_id,
        })
    }

    pub(super) async fn history(&self) -> Result<Vec<Event>, Box<dyn std::error::Error>> {
        Ok(self.store.read_history(&self.workflow_id).await?)
    }
}

pub(in super::super) const FIXED_RETRY_CONFIG: &str =
    r#"{"retry":{"max_attempts":3,"backoff":{"kind":"fixed","delay_ms":2}}}"#;