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
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160

use std::sync::Arc;

use aion_core::{ActivityId, RunId, WorkflowId, WorkflowStatus};
use aion_package::ContentHash;

use crate::registry::Registry;
use crate::registry::handle::{
    CompletionNotifier, HandleResidency, WorkflowHandle, WorkflowHandleParts,
};
use crate::runtime::config::RuntimeConfig;

use super::RuntimeHandle;

type TestResult = Result<(), Box<dyn std::error::Error>>;

fn live_handle(workflow_id: &WorkflowId, run_id: &RunId, pid: u64) -> WorkflowHandle {
    let store = Arc::new(aion_store::InMemoryStore::default());
    let recorder = crate::durability::Recorder::new(workflow_id.clone(), store);
    WorkflowHandle::new(WorkflowHandleParts {
        workflow_id: workflow_id.clone(),
        run_id: run_id.clone(),
        pid,
        workflow_type: "checkout".to_owned(),
        namespace: String::from("default"),
        loaded_version: ContentHash::from_bytes([1; 32]),
        cached_status: WorkflowStatus::Running,
        residency: HandleResidency::Resident,
        recorder,
        completion: CompletionNotifier::new(),
    })
}

#[test]
fn outbox_completion_lands_where_take_reads_it() -> TestResult {
    let runtime = RuntimeHandle::new(RuntimeConfig::new(None))?;
    let registry = Registry::default();
    let workflow_id = WorkflowId::new_v4();
    let run_id = RunId::new_v4();
    let pid = runtime.spawn_test_process()?;
    registry.insert(
        (workflow_id.clone(), run_id.clone()),
        live_handle(&workflow_id, &run_id, pid),
    )?;

    let ordinal = 3;
    let activity_id = ActivityId::from_sequence_position(ordinal);
    let delivered = runtime.deliver_outbox_completion(
        &registry,
        &workflow_id,
        &activity_id,
        None,
        r#"{"ok":true}"#.to_owned(),
    )?;

    assert!(delivered, "delivery to a live workflow must report true");
    let (payload, attempt) = runtime
        .take_activity_result(pid, ordinal)?
        .ok_or("completion was not retained where take_activity_result reads it")?;
    assert_eq!(attempt, None);
    assert_eq!(payload.bytes(), br#"{"ok":true}"#);

    let unknown = runtime.deliver_outbox_completion(
        &registry,
        &WorkflowId::new_v4(),
        &activity_id,
        None,
        "{}".to_owned(),
    )?;
    assert!(!unknown, "an unknown workflow must report not-live");

    runtime.shutdown()?;
    Ok(())
}

#[test]
fn outbox_completion_is_run_scoped_across_continue_as_new() -> TestResult {
    let runtime = RuntimeHandle::new(RuntimeConfig::new(None))?;
    let registry = Registry::default();
    let workflow_id = WorkflowId::new_v4();
    let prior_run = RunId::new_v4();
    let live_run = RunId::new_v4();
    let pid = runtime.spawn_test_process()?;
    registry.insert(
        (workflow_id.clone(), live_run.clone()),
        live_handle(&workflow_id, &live_run, pid),
    )?;

    let ordinal = 3;
    let activity_id = ActivityId::from_sequence_position(ordinal);
    let stale = runtime.deliver_outbox_completion(
        &registry,
        &workflow_id,
        &activity_id,
        Some(&prior_run),
        r#"{"from":"prior"}"#.to_owned(),
    )?;
    assert!(!stale, "a superseded run must not receive a completion");
    assert!(runtime.take_activity_result(pid, ordinal)?.is_none());

    let delivered = runtime.deliver_outbox_completion(
        &registry,
        &workflow_id,
        &activity_id,
        Some(&live_run),
        r#"{"from":"live"}"#.to_owned(),
    )?;
    assert!(delivered, "the live run must receive its completion");
    let (payload, attempt) = runtime
        .take_activity_result(pid, ordinal)?
        .ok_or("live-run completion was not retained")?;
    assert_eq!(attempt, None);
    assert_eq!(payload.bytes(), br#"{"from":"live"}"#);

    runtime.shutdown()?;
    Ok(())
}

#[test]
fn outbox_failure_is_run_scoped_across_continue_as_new() -> TestResult {
    let runtime = RuntimeHandle::new(RuntimeConfig::new(None))?;
    let registry = Registry::default();
    let workflow_id = WorkflowId::new_v4();
    let prior_run = RunId::new_v4();
    let live_run = RunId::new_v4();
    let pid = runtime.spawn_test_process()?;
    registry.insert(
        (workflow_id.clone(), live_run.clone()),
        live_handle(&workflow_id, &live_run, pid),
    )?;

    let ordinal = 5;
    let activity_id = ActivityId::from_sequence_position(ordinal);
    let stale = runtime.deliver_outbox_failure(
        &registry,
        &workflow_id,
        &activity_id,
        Some(&prior_run),
        "prior failed".to_owned(),
    )?;
    assert!(!stale, "a superseded run must not receive a failure");
    assert!(runtime.take_activity_error(pid, ordinal)?.is_none());

    let delivered = runtime.deliver_outbox_failure(
        &registry,
        &workflow_id,
        &activity_id,
        Some(&live_run),
        "live failed".to_owned(),
    )?;
    assert!(delivered, "the live run must receive its failure");
    assert!(runtime.take_activity_error(pid, ordinal)?.is_some());

    runtime.shutdown()?;
    Ok(())
}

#[path = "delivery_interleaving_tests.rs"]
mod interleavings;
#[path = "delivery_rollback_tests.rs"]
mod rollback;