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
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205

use super::*;

/// The live expiry decision must be a pure function of the RESOLUTION
/// snapshot. Race modeled: the await's snapshot (a stale read) lacks
/// the scope deadline's `TimerFired`, which is recorded by the time any
/// later read runs. Before the fix `expired_scope_message` re-read the
/// store, saw the fired deadline, and recorded the durable timeout
/// failure on the spot — a branch decided from events the resolution
/// never observed. After the fix the stale-snapshot pass suspends; the
/// deadline's wake re-enters with a fresh snapshot, records the timeout
/// failure durably, and a fresh engine epoch returns it verbatim while
/// appending nothing.
#[tokio::test(flavor = "multi_thread")]
async fn stale_snapshot_expiry_suspends_then_converges_with_replay() -> TestResult {
    // Stale snapshot = WorkflowStarted + Scheduled + Started: the
    // deadline `TimerFired` (seq 4) is the one event past the window.
    let backing = Arc::new(StaleReadStore::new(3));
    let store: Arc<dyn EventStore> = Arc::clone(&backing) as Arc<dyn EventStore>;
    let (workflow_id, run_id) = seed_pending_activity_then_deadline(&store, 7).await?;
    let harness =
        AwaitHarness::over_store(Arc::clone(&store), workflow_id.clone(), run_id.clone()).await?;
    harness.install_fresh_read_bridge();
    backing.set_stale_target(&workflow_id, 1);
    harness.arm_live_scope(7);

    // Pass 1 — stale resolution snapshot (no terminal, no TimerFired):
    // must suspend, never decide the branch from a fresh read; nothing
    // is recorded.
    assert_eq!(
        harness.step(),
        Ok(ActivityAwaitStep::Suspend),
        "a snapshot lacking both events must park, not branch"
    );
    assert_eq!(harness.history_len().await?, 4);

    // Pass 2 — fresh snapshot: the deadline is in the resolution read;
    // the timeout failure is recorded durably and returned.
    assert_eq!(
        harness.step(),
        Ok(ActivityAwaitStep::Failed(
            "timeout:deadline expired".to_owned()
        ))
    );
    let history = harness.store.read_history(&workflow_id).await?;
    assert!(
        matches!(history.last(), Some(Event::ActivityFailed { .. })),
        "the timeout branch must be recorded durably: {history:#?}"
    );
    let history_len = history.len();
    harness.shutdown()?;

    // Fresh engine epoch over the final store (the restart analogue),
    // scope replay-derived expired exactly as `arm_scope` derives it:
    // the recorded failure resolves verbatim, appending nothing.
    let replay = AwaitHarness::over_store(store, workflow_id, run_id).await?;
    replay.arm_replayed_expired_scope(7);
    assert_eq!(
        replay.step(),
        Ok(ActivityAwaitStep::Failed(
            "timeout:deadline expired".to_owned()
        )),
        "replay must take the same branch as the converged live run"
    );
    assert_eq!(replay.history_len().await?, history_len);
    replay.shutdown()
}

/// A completion sitting in the runtime maps settles the await — and is
/// recorded durably — ahead of the scope-expiry branch, even when the
/// resolution snapshot already contains the fired deadline. The
/// recorded terminal IS the decision, so a fresh engine epoch resolves
/// the completion identically (no deadline-vs-terminal seq ordering is
/// needed for activities: this await records its own terminals, no
/// third party races them into history).
#[tokio::test(flavor = "multi_thread")]
async fn delivered_completion_settles_durably_ahead_of_snapshot_expiry() -> TestResult {
    let backing = Arc::new(StaleReadStore::new(0));
    let store: Arc<dyn EventStore> = Arc::clone(&backing) as Arc<dyn EventStore>;
    let (workflow_id, run_id) = seed_pending_activity_then_deadline(&store, 7).await?;
    let harness =
        AwaitHarness::over_store(Arc::clone(&store), workflow_id.clone(), run_id.clone()).await?;
    harness.arm_live_scope(7);
    harness.runtime.deliver_activity_completion_message(
        harness.pid,
        "activity:0",
        r#""r0""#.to_owned(),
    )?;

    assert_eq!(
        harness.step(),
        Ok(ActivityAwaitStep::Completed(br#""r0""#.to_vec()))
    );
    let history = harness.store.read_history(&workflow_id).await?;
    assert!(
        matches!(history.last(), Some(Event::ActivityCompleted { .. })),
        "the completion must be recorded durably: {history:#?}"
    );
    let history_len = history.len();
    harness.shutdown()?;

    let replay = AwaitHarness::over_store(store, workflow_id, run_id).await?;
    replay.arm_replayed_expired_scope(7);
    assert_eq!(
        replay.step(),
        Ok(ActivityAwaitStep::Completed(br#""r0""#.to_vec())),
        "replay must resolve the recorded completion, not re-derive the race"
    );
    assert_eq!(replay.history_len().await?, history_len);
    replay.shutdown()
}

#[tokio::test(flavor = "multi_thread")]
async fn poisoned_take_fails_typed_and_monitor_drains_retained_state() -> TestResult {
    let backing = Arc::new(StaleReadStore::new(0));
    let store: Arc<dyn EventStore> = Arc::clone(&backing) as Arc<dyn EventStore>;
    let (workflow_id, run_id) = seed_pending_activity_then_deadline(&store, 7).await?;
    let harness = AwaitHarness::over_store(store, workflow_id, run_id).await?;
    let baseline = harness.runtime.activity_delivery_gate_count();
    harness
        .runtime
        .deliver_activity_completion_message_with_attempt(
            harness.pid,
            "activity:0",
            r#""retained""#.to_owned(),
            Some(3),
        )?;
    assert_eq!(harness.runtime.retained_activity_completions(), 1);
    assert_eq!(
        harness.runtime.retained_activity_attempt_count_for_test(),
        1
    );
    harness
        .runtime
        .force_activity_delivery_poison_for_test(harness.pid)?;

    let step = harness.step_typed();
    assert!(matches!(
        step,
        Err(EngineError::ActivityDeliveryPoisoned { process_id })
            if process_id == harness.pid
    ));
    assert_eq!(
        harness.history_len().await?,
        4,
        "poison must neither suspend nor record a fabricated attempt-one completion"
    );

    let (monitor_sender, monitor_receiver) = std::sync::mpsc::channel();
    harness
        .runtime
        .monitor_process_for_test(harness.pid, move |outcome| {
            if monitor_sender.send(outcome).is_err() {
                tracing::error!("poisoned-take monitor receiver dropped");
            }
        })?;
    harness.runtime.cancel_pid(harness.pid)?;
    let monitored = monitor_receiver.recv_timeout(std::time::Duration::from_secs(10))?;
    assert!(matches!(
        monitored,
        Err(EngineError::ActivityDeliveryPoisoned { process_id })
            if process_id == harness.pid
    ));
    assert_eq!(harness.runtime.retained_activity_completions(), 0);
    assert_eq!(
        harness.runtime.retained_activity_attempt_count_for_test(),
        0
    );
    assert_eq!(harness.runtime.activity_delivery_gate_count(), baseline);
    harness.shutdown()
}
/// The completion task notes the final attempt where the awaiting NIF
/// takes it, and the recorded terminal carries it (NOI-0 fidelity across
/// the retry loop).
#[tokio::test(flavor = "multi_thread")]
async fn awaited_terminal_records_the_noted_final_attempt() -> TestResult {
    let backing = Arc::new(StaleReadStore::new(0));
    let store: Arc<dyn EventStore> = Arc::clone(&backing) as Arc<dyn EventStore>;
    let (workflow_id, run_id) = seed_pending_activity_then_deadline(&store, 7).await?;
    let harness = AwaitHarness::over_store(Arc::clone(&store), workflow_id.clone(), run_id).await?;
    harness
        .runtime
        .deliver_activity_failure_message_with_attempt(
            harness.pid,
            "activity:0",
            "retryable:reset three".to_owned(),
            Some(3),
        )?;

    assert_eq!(
        harness.step(),
        Ok(ActivityAwaitStep::Failed(
            "retryable:reset three".to_owned()
        ))
    );
    let history = store.read_history(&workflow_id).await?;
    assert!(
        matches!(
            history.last(),
            Some(Event::ActivityFailed { attempt: 3, error, .. })
                if error.message == "retryable:reset three"
        ),
        "the recorded terminal must carry the noted final attempt: {history:#?}"
    );
    harness.shutdown()
}