orka 0.3.1

An asynchronous, pluggable, and type-safe workflow engine for Rust, designed for orchestrating complex multi-step business processes.
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

//! Tests for `resolve_plan` and the step-isolation runners `run_step`/`run_from`/
//! `run_until`, including their contractual trace shape (step events only, no
//! `RunStarted`/`RunFinished`).

mod common;

use common::{setup_tracing, TestContext, TestError};
use orka::test_util::noop_pipeline;
use orka::{
  ContextData, Pipeline, PipelineControl, PipelineResult, PlannedAction, SkipReason, StepPlan, TraceCollector,
  TraceEventKind,
};

fn recording_pipeline() -> Pipeline<TestContext, TestError> {
  let mut p: Pipeline<TestContext, TestError> = Pipeline::new(["load", "process", "notify"]);
  for name in ["load", "process", "notify"] {
    p.on_root(name, move |ctx| async move {
      ctx.write().steps_executed.push(name.to_string());
      Ok(PipelineControl::Continue)
    });
  }
  p
}

#[test]
fn resolve_plan_reports_run_skip_and_missing_per_seeded_context() {
  setup_tracing();
  let mut p: Pipeline<TestContext, TestError> = noop_pipeline(["always", "conditional", "empty_optional"]);
  p.clear_on("empty_optional");
  p.optional("empty_optional");
  p.skip_if("conditional", |ctx| ctx.read().counter > 10);

  // Table test over seeded contexts; nothing executes.
  let low = ContextData::new(TestContext {
    counter: 1,
    ..TestContext::default()
  });
  assert_eq!(
    p.resolve_plan(&low),
    vec![
      StepPlan {
        name: "always".into(),
        action: PlannedAction::Run
      },
      StepPlan {
        name: "conditional".into(),
        action: PlannedAction::Run
      },
      StepPlan {
        name: "empty_optional".into(),
        action: PlannedAction::Skip(SkipReason::OptionalWithoutHandlers)
      },
    ]
  );

  let high = ContextData::new(TestContext {
    counter: 42,
    ..TestContext::default()
  });
  assert_eq!(
    p.resolve_plan(&high)[1].action,
    PlannedAction::Skip(SkipReason::SkipCondition { label: None })
  );

  // A required, handler-less step is predicted as the failure `run` would produce.
  let mut broken: Pipeline<TestContext, TestError> = Pipeline::new(["nothing_here"]);
  broken.skip_if("nothing_here", |_ctx| false);
  assert_eq!(
    broken.resolve_plan(&ContextData::new(TestContext::default()))[0].action,
    PlannedAction::FailMissingHandlers
  );
}

#[test]
fn resolve_plan_executes_no_handlers() {
  setup_tracing();
  let p = recording_pipeline();
  let ctx = ContextData::new(TestContext::default());
  let plan = p.resolve_plan(&ctx);
  assert_eq!(plan.len(), 3);
  assert!(ctx.read().steps_executed.is_empty());
}

#[tokio::test]
async fn run_step_executes_exactly_one_step() {
  setup_tracing();
  let p = recording_pipeline();
  let ctx = ContextData::new(TestContext::default());
  let result = p.run_step("process", ctx.clone()).await.unwrap();
  assert_eq!(result, PipelineResult::Completed);
  assert_eq!(ctx.read().steps_executed, vec!["process"]);
}

#[tokio::test]
async fn run_step_still_respects_skip_if() {
  setup_tracing();
  let mut p = recording_pipeline();
  p.skip_if("process", |_ctx| true);
  let ctx = ContextData::new(TestContext::default());
  p.run_step("process", ctx.clone()).await.unwrap();
  assert!(ctx.read().steps_executed.is_empty());
}

#[tokio::test]
async fn run_from_and_run_until_are_inclusive_ranges() {
  setup_tracing();
  let p = recording_pipeline();

  let ctx = ContextData::new(TestContext::default());
  p.run_from("process", ctx.clone()).await.unwrap();
  assert_eq!(ctx.read().steps_executed, vec!["process", "notify"]);

  let ctx2 = ContextData::new(TestContext::default());
  p.run_until("process", ctx2.clone()).await.unwrap();
  assert_eq!(ctx2.read().steps_executed, vec!["load", "process"]);
}

#[tokio::test]
async fn partial_runners_error_on_unknown_step() {
  setup_tracing();
  let p = recording_pipeline();
  let ctx = ContextData::new(TestContext::default());
  for result in [
    p.run_step("nope", ctx.clone()).await,
    p.run_from("nope", ctx.clone()).await,
    p.run_until("nope", ctx.clone()).await,
  ] {
    let err = result.unwrap_err();
    assert!(matches!(&err, TestError::Orka(msg) if msg.contains("StepNotFound")), "got: {:?}", err);
  }
}

#[tokio::test]
async fn partial_run_trace_is_step_events_only() {
  setup_tracing();
  // Contractual: partial runs are not runs, so they emit no RunStarted/RunFinished, but
  // their step events are still run_id-tagged for for_run filtering.
  let p = recording_pipeline();
  let trace = TraceCollector::new();
  p.set_tracer(trace.clone());

  p.run_step("process", ContextData::new(TestContext::default())).await.unwrap();

  let events = trace.events();
  assert!(!events.is_empty());
  assert!(events.iter().all(|e| !matches!(
    e.kind,
    TraceEventKind::RunStarted | TraceEventKind::RunFinished { .. }
  )));
  assert_eq!(trace.run_count(), 0);

  let run_ids = trace.run_ids();
  assert_eq!(run_ids.len(), 1);
  let run = trace.for_run(run_ids[0]);
  assert_eq!(run.completed_steps(), vec!["process"]);

  // Step indexes reflect the position in the full pipeline, not the slice.
  assert!(events.iter().any(|e| matches!(
    e.kind,
    TraceEventKind::StepStarted { ref step, index: 1 } if step == "process"
  )));
}

#[test]
fn skip_if_labeled_shows_its_label_in_resolve_plan() {
  setup_tracing();
  let mut p: Pipeline<TestContext, TestError> = noop_pipeline(["drain", "install"]);
  p.skip_if_labeled("drain", "drain disabled by config", |ctx| ctx.read().counter == 0);

  let plan = p.resolve_plan(&ContextData::new(TestContext::default()));
  assert_eq!(
    plan[0].action,
    PlannedAction::Skip(SkipReason::SkipCondition {
      label: Some("drain disabled by config".to_string())
    })
  );

  // Re-registering an unlabeled skip_if clears the stale label; clear_skip_condition too.
  p.skip_if("drain", |_ctx| true);
  assert_eq!(
    p.resolve_plan(&ContextData::new(TestContext::default()))[0].action,
    PlannedAction::Skip(SkipReason::SkipCondition { label: None })
  );
}

#[test]
fn step_plan_renders_a_readable_preview() {
  setup_tracing();
  let mut p: Pipeline<TestContext, TestError> = noop_pipeline(["prepare", "drain", "cleanup", "install"]);
  p.skip_if_labeled("drain", "drain disabled by config", |_ctx| true);
  p.skip_if("cleanup", |_ctx| true);
  p.clear_on("install"); // required step left with no handlers

  let plan = p.resolve_plan(&ContextData::new(TestContext::default()));
  let rendered: Vec<String> = plan.iter().map(|step| step.to_string()).collect();
  assert_eq!(
    rendered,
    vec![
      "prepare: run",
      "drain: skip (drain disabled by config)",
      "cleanup: skip (skip_if condition)",
      "install: fail (required step has no handlers)",
    ]
  );
}