aion-rs 0.6.1

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

//! End-to-end engine cancellation integration tests over `InMemoryStore`.

mod common;

use std::thread;
use std::time::{Duration, Instant};

use aion::dispatch_activity;
use aion_core::Event;
use serde_json::json;

use common::{FIXTURE_MODULE, input_payload, payload};

fn wait_until_not_live(runtime: &aion::RuntimeHandle, pid: aion::Pid) -> bool {
    let deadline = Instant::now() + Duration::from_secs(2);
    while Instant::now() < deadline {
        if !runtime.is_live(pid) {
            return true;
        }
        thread::sleep(Duration::from_millis(10));
    }
    !runtime.is_live(pid)
}

#[tokio::test]
async fn cancel_records_event_deregisters_and_kills_workflow()
-> Result<(), Box<dyn std::error::Error>> {
    let (engine, store) = common::engine_with_fixture("wait").await?;
    let handle = engine
        .start_workflow(
            FIXTURE_MODULE,
            input_payload()?,
            std::collections::HashMap::new(),
        )
        .await?;

    let reason = "caller requested cancellation";
    engine
        .cancel(handle.workflow_id(), handle.run_id(), reason)
        .await?;

    assert!(wait_until_not_live(engine.runtime(), handle.pid()));
    assert!(
        engine
            .registry()
            .get(handle.workflow_id(), handle.run_id())?
            .is_none()
    );

    let history = store.read_history(handle.workflow_id()).await?;
    match history.last() {
        Some(Event::WorkflowCancelled {
            reason: recorded_reason,
            ..
        }) => assert_eq!(recorded_reason, reason),
        other => {
            return Err(format!("expected final WorkflowCancelled event, found {other:?}").into());
        }
    }

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

#[tokio::test]
async fn cancel_propagates_kill_to_linked_activity() -> Result<(), Box<dyn std::error::Error>> {
    let (engine, _store) = common::engine_with_fixture("wait").await?;
    let handle = engine
        .start_workflow(
            FIXTURE_MODULE,
            input_payload()?,
            std::collections::HashMap::new(),
        )
        .await?;
    let deployed_module = engine
        .workflow_catalog()
        .routed(FIXTURE_MODULE)?
        .ok_or("fixture workflow was not loaded")?
        .deployed_entry_module()
        .to_owned();
    let activity = dispatch_activity(
        engine.runtime(),
        handle.pid(),
        &deployed_module,
        "activity",
        &payload(&json!(null))?,
    )?;

    assert!(engine.runtime().is_live(activity));

    let reason = "caller requested cancellation";
    engine
        .cancel(handle.workflow_id(), handle.run_id(), reason)
        .await?;

    assert!(wait_until_not_live(engine.runtime(), handle.pid()));
    assert!(wait_until_not_live(engine.runtime(), activity));

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