1use crate::criterion;
9use crate::expression::{self, Exchange, ExpressionError, Scope, StepState, WorkflowState};
10use crate::http::{HttpRequest, HttpResponse};
11use crate::operation::{self, Source};
12use crate::report::{
13 CriterionOutcome, ExecutionError, ExecutionReport, Outcome, Performed, StepRecord,
14};
15use crate::select;
16use crate::select::SelectError;
17use roas_arazzo::v1_1::{
18 Criterion, CriterionKind, CriterionType, Description, FailureActionType, Parameter,
19 ParameterLocation, ReusableOr, SourceType, Step, SuccessActionType, ValueOrSelector, Workflow,
20};
21use serde_json::{Map, Value};
22use std::collections::{BTreeMap, BTreeSet};
23use std::time::{Duration, Instant};
24
25#[derive(Clone, Copy, Debug)]
27struct Limits {
28 steps: usize,
29 depth: usize,
30 retries: u32,
31}
32
33impl Default for Limits {
34 fn default() -> Self {
35 Self {
36 steps: 1_000,
37 depth: 8,
38 retries: 10,
39 }
40 }
41}
42
43#[derive(Clone, Debug, Default)]
47pub struct Options {
48 workflow: Option<String>,
49 inputs: Map<String, Value>,
50 sources: BTreeMap<String, Source>,
51 base_urls: BTreeMap<String, String>,
52 headers: Vec<(String, String)>,
53 limits: Limits,
54}
55
56impl Options {
57 #[must_use]
60 pub fn new() -> Self {
61 Self::default()
62 }
63
64 #[must_use]
66 pub fn workflow(mut self, workflow_id: impl Into<String>) -> Self {
67 self.workflow = Some(workflow_id.into());
68 self
69 }
70
71 #[must_use]
73 pub fn input(mut self, name: impl Into<String>, value: impl Into<Value>) -> Self {
74 self.inputs.insert(name.into(), value.into());
75 self
76 }
77
78 #[must_use]
80 pub fn inputs(mut self, inputs: Value) -> Self {
81 if let Value::Object(inputs) = inputs {
82 self.inputs = inputs;
83 }
84 self
85 }
86
87 #[must_use]
94 pub fn source(
95 mut self,
96 name: impl Into<String>,
97 url: impl Into<String>,
98 document: Value,
99 ) -> Self {
100 self.sources.insert(
101 name.into(),
102 Source {
103 url: url.into(),
104 document,
105 },
106 );
107 self
108 }
109
110 #[must_use]
113 pub fn base_url(mut self, source_name: impl Into<String>, url: impl Into<String>) -> Self {
114 self.base_urls.insert(source_name.into(), url.into());
115 self
116 }
117
118 #[must_use]
120 pub fn header(mut self, name: impl Into<String>, value: impl Into<String>) -> Self {
121 self.headers.push((name.into(), value.into()));
122 self
123 }
124
125 #[must_use]
127 pub fn max_steps(mut self, steps: usize) -> Self {
128 self.limits.steps = steps;
129 self
130 }
131
132 #[must_use]
134 pub fn max_depth(mut self, depth: usize) -> Self {
135 self.limits.depth = depth;
136 self
137 }
138
139 #[must_use]
141 pub fn max_retries(mut self, retries: u32) -> Self {
142 self.limits.retries = retries;
143 self
144 }
145}
146
147#[derive(Debug)]
154pub enum Progress {
155 Send(HttpRequest),
158 Wait(Duration),
160 Done(Box<ExecutionReport>),
162}
163
164fn scope<'s>(
166 frame: &'s Frame<'_>,
167 steps: &'s BTreeMap<String, StepState>,
168 here: Option<&'s Exchange>,
169 finished: &'s BTreeMap<String, WorkflowState>,
170 ambient: &'s Ambient,
171) -> Scope<'s> {
172 Scope {
173 inputs: &frame.inputs,
174 outputs: &frame.outputs,
175 steps,
176 workflows: finished,
177 here,
178 sources: &ambient.sources,
179 components: &ambient.components,
180 self_: ambient.self_.as_deref(),
181 declared_steps: &frame.declared,
182 declared_workflows: &ambient.workflows,
183 }
184}
185
186#[derive(Clone, Copy, Debug, PartialEq, Eq)]
188enum Then {
189 Advance,
191 Retry,
193 EndCaller,
195}
196
197struct Frame<'d> {
199 workflow: &'d Workflow,
200 inputs: Value,
201 order: Vec<usize>,
203 at: usize,
204 steps: BTreeMap<String, StepState>,
205 outputs: BTreeMap<String, Value>,
206 attempts: BTreeMap<String, u32>,
209 declared: BTreeSet<String>,
211 retries: BTreeMap<(String, usize), u32>,
215 caller: Option<(String, Then)>,
218 started: Instant,
220 detour: Option<usize>,
223 outcome: Outcome,
224}
225
226struct Completion {
228 exchange: Option<Exchange>,
230 given: BTreeMap<String, Value>,
233 default_pass: bool,
235 attempt: u32,
236 performed: Performed,
237 elapsed: Duration,
238}
239
240struct Pending {
242 step: usize,
243 attempt: u32,
244 exchange: Exchange,
245 started: Instant,
246}
247
248struct Ambient {
251 sources: Value,
253 components: Value,
255 self_: Option<String>,
257 workflows: BTreeSet<String>,
259}
260
261pub struct Run<'d> {
263 description: &'d Description,
264 options: &'d Options,
265 ambient: Ambient,
266 frames: Vec<Frame<'d>>,
267 queue: Vec<&'d Workflow>,
270 finished: BTreeMap<String, WorkflowState>,
271 pending: Option<Pending>,
272 wait: Option<Duration>,
273 records: Vec<StepRecord>,
274 unsupplied: Vec<String>,
278 taken: usize,
279 report: Option<Box<ExecutionReport>>,
280}
281
282impl<'d> Run<'d> {
283 pub fn start(
292 description: &'d Description,
293 options: &'d Options,
294 ) -> Result<Self, ExecutionError> {
295 let wanted = match &options.workflow {
296 Some(id) => description
297 .workflows
298 .iter()
299 .find(|workflow| &workflow.workflow_id == id)
300 .ok_or_else(|| ExecutionError::UnknownWorkflow(id.clone()))?,
301 None => description
302 .workflows
303 .first()
304 .ok_or_else(|| ExecutionError::UnknownWorkflow(String::new()))?,
305 };
306
307 let sources = serde_json::to_value(
308 description
309 .source_descriptions
310 .iter()
311 .map(|source| (source.name.clone(), source))
312 .collect::<BTreeMap<_, _>>(),
313 )
314 .unwrap_or(Value::Null);
315 let components = description
316 .components
317 .as_ref()
318 .and_then(|components| serde_json::to_value(components).ok())
319 .unwrap_or(Value::Null);
320
321 let mut queue = ordered_workflows(description, wanted)?;
322 let first = queue.remove(0);
323 let mut run = Self {
324 description,
325 options,
326 ambient: Ambient {
327 sources,
328 components,
329 self_: description.self_.clone(),
330 workflows: description
331 .workflows
332 .iter()
333 .map(|workflow| workflow.workflow_id.clone())
334 .collect(),
335 },
336 frames: Vec::new(),
337 queue,
338 finished: BTreeMap::new(),
339 pending: None,
340 wait: None,
341 records: Vec::new(),
342 unsupplied: description
343 .source_descriptions
344 .iter()
345 .filter(|source| source.type_ != Some(SourceType::Arazzo))
348 .filter(|source| !options.sources.contains_key(&source.name))
349 .map(|source| source.name.clone())
350 .collect(),
351 taken: 0,
352 report: None,
353 };
354 run.enter(first, Value::Object(options.inputs.clone()), None)?;
355 Ok(run)
356 }
357
358 pub fn advance(&mut self) -> Result<Progress, ExecutionError> {
368 if let Some(wait) = self.wait.take() {
369 return Ok(Progress::Wait(wait));
370 }
371 if let Some(pending) = &self.pending {
375 return Err(ExecutionError::Awaiting {
376 method: pending.exchange.request.method.clone(),
377 url: pending.exchange.request.url.clone(),
378 });
379 }
380 loop {
381 if let Some(report) = self.report.take() {
382 return Ok(Progress::Done(report));
383 }
384 let Some(frame) = self.frames.last() else {
385 return Ok(Progress::Done(Box::new(self.finish())));
386 };
387 if frame.at >= frame.order.len() {
389 self.leave()?;
390 continue;
391 }
392
393 self.taken += 1;
394 if self.taken > self.options.limits.steps {
395 return Err(ExecutionError::Limit {
396 limit: "step",
397 at: self.options.limits.steps,
398 });
399 }
400
401 let index = frame.order[frame.at];
402 let step = &frame.workflow.steps[index];
403 if step.workflow_id.is_some() {
404 self.call(index)?;
405 continue;
406 }
407
408 let (request, exchange) = self.build(index)?;
409 let attempt = self
410 .frames
411 .last()
412 .and_then(|frame| frame.attempts.get(&step.step_id).copied())
413 .unwrap_or(0)
414 + 1;
415 self.pending = Some(Pending {
416 step: index,
417 attempt,
418 exchange,
419 started: Instant::now(),
420 });
421 return Ok(Progress::Send(request));
422 }
423 }
424
425 pub fn supply(&mut self, response: HttpResponse) -> Result<(), ExecutionError> {
433 let Some(mut pending) = self.pending.take() else {
434 return Err(ExecutionError::NotWaiting);
435 };
436 let elapsed = pending.started.elapsed();
437 let status = response.status;
438 pending.exchange.response_body = response.body_as_json();
439 pending.exchange.response = Some(response);
440
441 if self.frames.last().is_none() {
442 return Err(ExecutionError::NotWaiting);
443 }
444 let performed = Performed::Request {
445 method: pending.exchange.request.method.clone(),
446 url: pending.exchange.request.url.clone(),
447 status,
448 };
449 self.complete(
451 pending.step,
452 Completion {
453 exchange: Some(pending.exchange),
454 given: BTreeMap::new(),
455 default_pass: (200..400).contains(&status),
456 attempt: pending.attempt,
457 performed,
458 elapsed,
459 },
460 )
461 }
462
463 fn complete(&mut self, index: usize, done: Completion) -> Result<(), ExecutionError> {
472 let frame = self.frames.last().expect("a frame to complete in");
473 let step = &frame.workflow.steps[index];
474 let step_id = step.step_id.clone();
475 let workflow_id = frame.workflow.workflow_id.clone();
476
477 let mut state = StepState {
480 exchange: done.exchange.clone(),
481 outputs: done.given.clone(),
482 passed: true,
483 };
484 let (passed, criteria, outputs) = {
485 let mut ahead = frame.steps.clone();
486 ahead.insert(step_id.clone(), state.clone());
487 let scope = scope(
488 frame,
489 &ahead,
490 done.exchange.as_ref(),
491 &self.finished,
492 &self.ambient,
493 );
494
495 let mut criteria = Vec::with_capacity(step.success_criteria.len());
496 let mut passed = if step.success_criteria.is_empty() {
499 done.default_pass
500 } else {
501 true
502 };
503 for criterion in &step.success_criteria {
504 let holds = criterion::passes(criterion, &scope)?;
505 criteria.push(CriterionOutcome {
506 condition: criterion.condition.clone(),
507 passed: holds,
508 });
509 passed = passed && holds;
510 }
511 let outputs = if passed {
515 let mut outputs = done.given.clone();
516 outputs.extend(evaluate_outputs(&step.outputs, &scope)?);
517 outputs
518 } else {
519 BTreeMap::new()
524 };
525 (passed, criteria, outputs)
526 };
527 state.outputs = outputs.clone();
528 state.passed = passed;
529
530 let frame = self.frames.last_mut().expect("the frame is still there");
534 frame.steps.insert(step_id.clone(), state);
535
536 let action = self.decide(index, passed, done.exchange.as_ref())?;
537 let described = describe(&action);
538
539 self.records.push(StepRecord {
540 workflow_id,
541 step_id,
542 attempt: done.attempt,
543 performed: done.performed,
544 criteria,
545 passed,
546 outputs,
547 action: described,
548 elapsed: done.elapsed,
549 });
550 self.apply(action)
551 }
552
553 fn enter(
555 &mut self,
556 workflow: &'d Workflow,
557 inputs: Value,
558 caller: Option<(String, Then)>,
559 ) -> Result<(), ExecutionError> {
560 if self.frames.len() >= self.options.limits.depth {
561 return Err(ExecutionError::Limit {
562 limit: "workflow depth",
563 at: self.options.limits.depth,
564 });
565 }
566 self.frames.push(Frame {
567 workflow,
568 inputs,
569 order: ordered_steps(workflow, self.description)?,
570 at: 0,
571 steps: BTreeMap::new(),
572 outputs: BTreeMap::new(),
573 declared: workflow
574 .steps
575 .iter()
576 .map(|step| step.step_id.clone())
577 .collect(),
578 attempts: BTreeMap::new(),
579 retries: BTreeMap::new(),
580 caller,
581 started: Instant::now(),
582 detour: None,
583 outcome: Outcome::Succeeded,
584 });
585 Ok(())
586 }
587
588 fn leave(&mut self) -> Result<(), ExecutionError> {
591 let frame = self.frames.pop().expect("a frame to leave");
592 let outputs = {
593 let scope = scope(&frame, &frame.steps, None, &self.finished, &self.ambient);
594 if frame.outcome == Outcome::Succeeded {
595 evaluate_outputs(&frame.workflow.outputs, &scope)?
596 } else {
597 evaluate_what_ran(&frame.workflow.outputs, &scope)?
601 }
602 };
603 self.finished.insert(
604 frame.workflow.workflow_id.clone(),
605 WorkflowState {
606 inputs: frame.inputs.clone(),
607 outputs: outputs.clone(),
608 },
609 );
610
611 let Some((step_id, then)) = frame.caller else {
612 if self.queue.is_empty() {
615 self.report = Some(Box::new(ExecutionReport {
616 workflow_id: frame.workflow.workflow_id.clone(),
617 outcome: frame.outcome,
618 outputs,
619 steps: std::mem::take(&mut self.records),
620 }));
621 } else {
622 let next = self.queue.remove(0);
623 let inputs = Value::Object(self.options.inputs.clone());
624 self.enter(next, inputs, None)?;
625 }
626 return Ok(());
627 };
628 let Some(parent) = self.frames.last() else {
629 return Ok(());
630 };
631 let index = parent.order[parent.at];
632 debug_assert_eq!(parent.workflow.steps[index].step_id, step_id);
633
634 match then {
635 Then::EndCaller => {
639 let parent = self.frames.last_mut().expect("the parent is still there");
640 parent.steps.insert(
641 step_id,
642 StepState {
643 exchange: None,
644 outputs,
645 passed: frame.outcome != Outcome::Failed,
646 },
647 );
648 parent.at = parent.order.len();
649 parent.outcome = frame.outcome;
650 Ok(())
651 }
652 Then::Retry => Ok(()),
655 Then::Advance => {
658 let elapsed = frame.started.elapsed();
659 let timed_out = parent.workflow.steps[index]
660 .timeout
661 .and_then(|timeout| u64::try_from(timeout).ok())
662 .is_some_and(|timeout| elapsed > Duration::from_millis(timeout));
663 let attempt = self
664 .frames
665 .last()
666 .and_then(|parent| parent.attempts.get(&step_id).copied())
667 .unwrap_or(0)
668 + 1;
669 self.complete(
670 index,
671 Completion {
672 exchange: None,
673 given: outputs,
674 default_pass: frame.outcome != Outcome::Failed && !timed_out,
675 attempt,
676 performed: Performed::Workflow {
677 workflow_id: frame.workflow.workflow_id.clone(),
678 outcome: frame.outcome,
679 },
680 elapsed,
681 },
682 )
683 }
684 }
685 }
686
687 fn call(&mut self, index: usize) -> Result<(), ExecutionError> {
689 let frame = self.frames.last().expect("a frame to call from");
690 let step = &frame.workflow.steps[index];
691 let step_id = step.step_id.clone();
692 let wanted = step.workflow_id.clone().unwrap_or_default();
693 if wanted.starts_with('$') {
694 return Err(ExecutionError::Unsupported(format!(
695 "step `{step_id}` calls `{wanted}`, and this executor runs only workflows of the description it was given"
696 )));
697 }
698 let workflow = self
699 .description
700 .workflows
701 .iter()
702 .find(|workflow| workflow.workflow_id == wanted)
703 .ok_or_else(|| ExecutionError::UnknownWorkflow(wanted.clone()))?;
704
705 let mut arguments = frame.workflow.parameters.clone();
712 arguments.extend(step.parameters.iter().cloned());
713 let inputs = self.arguments(&arguments)?;
714 self.enter(workflow, inputs, Some((step_id, Then::Advance)))
715 }
716
717 fn arguments(&self, arguments: &[ReusableOr<Parameter>]) -> Result<Value, ExecutionError> {
720 let frame = self.frames.last().expect("a frame to pass arguments from");
721 let scope = scope(frame, &frame.steps, None, &self.finished, &self.ambient);
722 let mut inputs = Map::new();
727 for parameter in parameters(arguments, self.description, &scope)? {
728 inputs.insert(parameter.name, parameter.value);
729 }
730 Ok(Value::Object(inputs))
731 }
732
733 fn build(&self, index: usize) -> Result<(HttpRequest, Exchange), ExecutionError> {
735 let frame = self.frames.last().expect("a frame to build in");
736 let step = &frame.workflow.steps[index];
737 let endpoint = operation::resolve(
738 step,
739 &self.options.sources,
740 &self.options.base_urls,
741 &self.unsupplied,
742 )?;
743 let scope = scope(frame, &frame.steps, None, &self.finished, &self.ambient);
744
745 let mut resolved = parameters(&frame.workflow.parameters, self.description, &scope)?;
747 for parameter in parameters(&step.parameters, self.description, &scope)? {
748 resolved.retain(|existing| {
749 !(existing.name == parameter.name && existing.location == parameter.location)
750 });
751 resolved.push(parameter);
752 }
753
754 let mut path = BTreeMap::new();
755 let mut query = BTreeMap::new();
756 let mut headers: Vec<(String, String)> = Vec::new();
757 let mut cookies = Vec::new();
758 let mut querystring = None;
759 for parameter in resolved {
760 match parameter.location {
761 ParameterLocation::Path => {
762 path.insert(parameter.name, parameter.value);
763 }
764 ParameterLocation::Query => {
765 query.insert(parameter.name, parameter.value);
766 }
767 ParameterLocation::Querystring => {
768 querystring = Some(text(¶meter.value));
769 }
770 ParameterLocation::Header => {
771 headers.push((parameter.name, text(¶meter.value)));
772 }
773 ParameterLocation::Cookie => {
774 cookies.push(format!("{}={}", parameter.name, text(¶meter.value)));
775 }
776 ParameterLocation::Channel => {
777 return Err(ExecutionError::Unsupported(format!(
778 "step `{}` has a `channel` parameter, which belongs to an AsyncAPI step",
779 step.step_id
780 )));
781 }
782 }
783 }
784 if !cookies.is_empty() {
785 headers.push(("Cookie".to_owned(), cookies.join("; ")));
786 }
787 for (name, value) in &self.options.headers {
788 if !headers
789 .iter()
790 .any(|(existing, _)| existing.eq_ignore_ascii_case(name))
791 {
792 headers.push((name.clone(), value.clone()));
793 }
794 }
795
796 let body = body(step, &scope)?;
797 if let Some(body) = &body
798 && !headers
799 .iter()
800 .any(|(name, _)| name.eq_ignore_ascii_case("content-type"))
801 {
802 headers.push(("Content-Type".to_owned(), body.content_type.clone()));
803 }
804
805 let url = url(&endpoint, &path, &query, querystring.as_deref()).map_err(|reason| {
806 ExecutionError::BadRequest {
807 step: step.step_id.clone(),
808 reason,
809 }
810 })?;
811 let request = HttpRequest {
812 method: endpoint.method,
813 url,
814 headers,
815 body: body.as_ref().map(|body| body.bytes.clone()),
816 timeout: step
817 .timeout
818 .and_then(|timeout| u64::try_from(timeout).ok())
819 .map(Duration::from_millis),
820 };
821 Ok((
822 request.clone(),
823 Exchange {
824 request,
825 path,
826 query,
827 body: body.map(|body| body.value),
828 response: None,
829 response_body: None,
830 },
831 ))
832 }
833
834 fn decide(
839 &mut self,
840 index: usize,
841 passed: bool,
842 exchange: Option<&Exchange>,
843 ) -> Result<Action, ExecutionError> {
844 let frame = self.frames.last().expect("a frame to decide in");
845 let step = &frame.workflow.steps[index];
846 let scope = scope(frame, &frame.steps, exchange, &self.finished, &self.ambient);
847
848 if passed {
849 let actions = step
851 .on_success
852 .iter()
853 .chain(frame.workflow.success_actions.iter());
854 for action in actions {
855 let action = success_action(action, self.description)?;
856 if !holds(&action.criteria, &scope)? {
857 continue;
858 }
859 return Ok(match action.type_ {
860 SuccessActionType::End => Action::End(Outcome::Ended),
861 SuccessActionType::Goto => Action::Goto {
862 step: action.step_id.clone(),
863 workflow: action.workflow_id.clone(),
864 parameters: action.parameters.clone(),
865 },
866 });
867 }
868 return Ok(Action::Advance);
869 }
870
871 let actions = step
872 .on_failure
873 .iter()
874 .chain(frame.workflow.failure_actions.iter());
875 for (at, action) in actions.enumerate() {
876 let action = failure_action(action, self.description)?;
877 if !holds(&action.criteria, &scope)? {
878 continue;
879 }
880 return Ok(match action.type_ {
881 FailureActionType::End => Action::End(Outcome::Failed),
882 FailureActionType::Goto => Action::Goto {
883 step: action.step_id.clone(),
884 workflow: action.workflow_id.clone(),
885 parameters: action.parameters.clone(),
886 },
887 FailureActionType::Retry => {
888 let allowed = action
896 .retry_limit
897 .map_or(1, |limit| u32::try_from(limit).unwrap_or(u32::MAX));
898 let spent = frame
899 .retries
900 .get(&(step.step_id.clone(), at))
901 .copied()
902 .unwrap_or(0);
903 let taken = frame.attempts.get(&step.step_id).copied().unwrap_or(0);
908 if spent >= allowed || taken >= self.options.limits.retries {
909 continue;
910 }
911 Action::Retry {
912 at,
913 after: action.retry_after,
914 step: action.step_id.clone(),
915 workflow: action.workflow_id.clone(),
916 parameters: action.parameters.clone(),
917 }
918 }
919 });
920 }
921 Ok(Action::End(Outcome::Failed))
924 }
925
926 fn apply(&mut self, action: Action) -> Result<(), ExecutionError> {
928 let frame = self.frames.last_mut().expect("a frame to act in");
929 match action {
930 Action::Advance => {
931 match frame.detour.take() {
934 Some(back) => frame.at = back,
935 None => frame.at += 1,
936 }
937 Ok(())
938 }
939 Action::End(outcome) => {
940 frame.outcome = outcome;
941 frame.at = frame.order.len();
942 Ok(())
947 }
948 Action::Retry {
949 at,
950 after,
951 step: target,
952 workflow,
953 parameters: arguments,
954 } => {
955 let index = frame.order[frame.at];
959 let step_id = frame.workflow.steps[index].step_id.clone();
960 *frame.attempts.entry(step_id.clone()).or_insert(0) += 1;
961 *frame.retries.entry((step_id.clone(), at)).or_insert(0) += 1;
962 if let Some(after) = after.filter(|after| *after > 0.0) {
963 self.wait = Some(Duration::from_secs_f64(after));
964 }
965 match (target, workflow) {
966 (Some(target), _) => {
970 let at = position_of(frame, &target)?;
971 frame.detour = Some(frame.at);
972 frame.at = at;
973 Ok(())
974 }
975 (None, Some(workflow_id)) => {
977 let workflow = self
978 .description
979 .workflows
980 .iter()
981 .find(|workflow| workflow.workflow_id == workflow_id)
982 .ok_or_else(|| ExecutionError::UnknownWorkflow(workflow_id.clone()))?;
983 let inputs = self.arguments(&arguments)?;
984 self.enter(workflow, inputs, Some((step_id, Then::Retry)))
985 }
986 (None, None) => Ok(()),
988 }
989 }
990 Action::Goto {
991 step: Some(step_id),
992 ..
993 } => {
994 frame.at = position_of(frame, &step_id)?;
995 Ok(())
996 }
997 Action::Goto {
998 workflow: Some(workflow_id),
999 parameters: arguments,
1000 ..
1001 } => {
1002 let index = frame.order[frame.at];
1003 let step_id = frame.workflow.steps[index].step_id.clone();
1004 let workflow = self
1005 .description
1006 .workflows
1007 .iter()
1008 .find(|workflow| workflow.workflow_id == workflow_id)
1009 .ok_or_else(|| ExecutionError::UnknownWorkflow(workflow_id.clone()))?;
1010 let inputs = self.arguments(&arguments)?;
1011 self.enter(workflow, inputs, Some((step_id, Then::EndCaller)))
1012 }
1013 Action::Goto { .. } => Ok(()),
1014 }
1015 }
1016
1017 fn finish(&mut self) -> ExecutionReport {
1019 ExecutionReport {
1020 workflow_id: self
1021 .options
1022 .workflow
1023 .clone()
1024 .or_else(|| {
1025 self.description
1026 .workflows
1027 .first()
1028 .map(|workflow| workflow.workflow_id.clone())
1029 })
1030 .unwrap_or_default(),
1031 outcome: Outcome::Succeeded,
1032 outputs: BTreeMap::new(),
1033 steps: std::mem::take(&mut self.records),
1034 }
1035 }
1036}
1037
1038fn position_of(frame: &Frame<'_>, step_id: &str) -> Result<usize, ExecutionError> {
1040 let index = frame
1041 .workflow
1042 .steps
1043 .iter()
1044 .position(|step| step.step_id == step_id)
1045 .ok_or_else(|| ExecutionError::UnknownStep {
1046 workflow: frame.workflow.workflow_id.clone(),
1047 step: step_id.to_owned(),
1048 })?;
1049 Ok(frame
1050 .order
1051 .iter()
1052 .position(|&candidate| candidate == index)
1053 .unwrap_or(frame.order.len()))
1054}
1055
1056#[derive(Clone, Debug)]
1058enum Action {
1059 Advance,
1060 End(Outcome),
1061 Retry {
1062 at: usize,
1065 after: Option<f64>,
1066 step: Option<String>,
1068 workflow: Option<String>,
1070 parameters: Vec<ReusableOr<Parameter>>,
1071 },
1072 Goto {
1073 step: Option<String>,
1074 workflow: Option<String>,
1075 parameters: Vec<ReusableOr<Parameter>>,
1076 },
1077}
1078
1079fn describe(action: &Action) -> Option<String> {
1080 match action {
1081 Action::Advance => None,
1082 Action::End(Outcome::Failed) => Some("ended, failed".to_owned()),
1083 Action::End(_) => Some("ended".to_owned()),
1084 Action::Retry {
1085 step: Some(step), ..
1086 } => Some(format!("retry via step `{step}`")),
1087 Action::Retry {
1088 workflow: Some(workflow),
1089 ..
1090 } => Some(format!("retry via workflow `{workflow}`")),
1091 Action::Retry { .. } => Some("retry".to_owned()),
1092 Action::Goto {
1093 step: Some(step), ..
1094 } => Some(format!("goto step `{step}`")),
1095 Action::Goto {
1096 workflow: Some(workflow),
1097 ..
1098 } => Some(format!("goto workflow `{workflow}`")),
1099 Action::Goto { .. } => None,
1100 }
1101}
1102
1103struct Resolved {
1105 name: String,
1106 location: ParameterLocation,
1107 value: Value,
1108}
1109
1110fn parameters(
1113 list: &[ReusableOr<Parameter>],
1114 description: &Description,
1115 scope: &Scope<'_>,
1116) -> Result<Vec<Resolved>, ExecutionError> {
1117 let mut resolved = Vec::with_capacity(list.len());
1118 for entry in list {
1119 let (parameter, overridden) = match entry {
1120 ReusableOr::Item(parameter) => (parameter.clone(), None),
1121 ReusableOr::Reusable(reusable) => {
1122 let name = reusable
1123 .reference
1124 .strip_prefix("$components.parameters.")
1125 .ok_or_else(|| {
1126 ExecutionError::Unsupported(format!(
1127 "`{}` is not a component this executor can follow",
1128 reusable.reference
1129 ))
1130 })?;
1131 let parameter = description
1132 .components
1133 .as_ref()
1134 .and_then(|components| components.parameters.get(name))
1135 .ok_or_else(|| {
1136 ExecutionError::Unsupported(format!(
1137 "`{}` names a component the description has not got",
1138 reusable.reference
1139 ))
1140 })?;
1141 (parameter.clone(), reusable.value.clone())
1142 }
1143 };
1144 let value = match overridden {
1145 Some(value) => select::resolve(&value, scope)?,
1146 None => select::value_of(¶meter.value, scope)?,
1147 };
1148 resolved.push(Resolved {
1149 name: parameter.name,
1150 location: parameter.in_.unwrap_or(ParameterLocation::Query),
1151 value,
1152 });
1153 }
1154 Ok(resolved)
1155}
1156
1157fn success_action(
1159 entry: &ReusableOr<roas_arazzo::v1_1::SuccessAction>,
1160 description: &Description,
1161) -> Result<roas_arazzo::v1_1::SuccessAction, ExecutionError> {
1162 match entry {
1163 ReusableOr::Item(action) => Ok(action.clone()),
1164 ReusableOr::Reusable(reusable) => reusable
1165 .reference
1166 .strip_prefix("$components.successActions.")
1167 .and_then(|name| {
1168 description
1169 .components
1170 .as_ref()
1171 .and_then(|components| components.success_actions.get(name))
1172 })
1173 .cloned()
1174 .ok_or_else(|| {
1175 ExecutionError::Unsupported(format!(
1176 "`{}` names a component the description has not got",
1177 reusable.reference
1178 ))
1179 }),
1180 }
1181}
1182
1183fn failure_action(
1185 entry: &ReusableOr<roas_arazzo::v1_1::FailureAction>,
1186 description: &Description,
1187) -> Result<roas_arazzo::v1_1::FailureAction, ExecutionError> {
1188 match entry {
1189 ReusableOr::Item(action) => Ok(action.clone()),
1190 ReusableOr::Reusable(reusable) => reusable
1191 .reference
1192 .strip_prefix("$components.failureActions.")
1193 .and_then(|name| {
1194 description
1195 .components
1196 .as_ref()
1197 .and_then(|components| components.failure_actions.get(name))
1198 })
1199 .cloned()
1200 .ok_or_else(|| {
1201 ExecutionError::Unsupported(format!(
1202 "`{}` names a component the description has not got",
1203 reusable.reference
1204 ))
1205 }),
1206 }
1207}
1208
1209fn holds(criteria: &[Criterion], scope: &Scope<'_>) -> Result<bool, ExecutionError> {
1212 for criterion in criteria {
1213 if !criterion::passes(criterion, scope)? {
1214 return Ok(false);
1215 }
1216 }
1217 Ok(true)
1218}
1219
1220fn evaluate_what_ran(
1230 outputs: &BTreeMap<String, ValueOrSelector>,
1231 scope: &Scope<'_>,
1232) -> Result<BTreeMap<String, Value>, ExecutionError> {
1233 let mut named = BTreeMap::new();
1234 for (name, value) in outputs {
1235 match select::value_of(value, scope) {
1236 Ok(value) => {
1237 named.insert(name.clone(), value);
1238 }
1239 Err(SelectError::Expression(ExpressionError::NotRun { .. })) => {}
1240 Err(error) => return Err(error.into()),
1241 }
1242 }
1243 Ok(named)
1244}
1245
1246fn evaluate_outputs(
1248 outputs: &BTreeMap<String, ValueOrSelector>,
1249 scope: &Scope<'_>,
1250) -> Result<BTreeMap<String, Value>, ExecutionError> {
1251 let mut resolved = BTreeMap::new();
1252 for (name, value) in outputs {
1253 resolved.insert(name.clone(), select::value_of(value, scope)?);
1254 }
1255 Ok(resolved)
1256}
1257
1258struct Body {
1261 bytes: Vec<u8>,
1262 value: Value,
1263 content_type: String,
1264}
1265
1266fn body(step: &Step, scope: &Scope<'_>) -> Result<Option<Body>, ExecutionError> {
1268 let Some(request_body) = &step.request_body else {
1269 return Ok(None);
1270 };
1271 let mut payload = match &request_body.payload {
1272 Some(payload) => select::resolve(payload, scope)?,
1273 None => Value::Null,
1274 };
1275 for replacement in &request_body.replacements {
1276 let value = select::value_of(&replacement.value, scope)?;
1277 let language = match &replacement.target_selector_type {
1278 Some(type_) => select::kind_of(type_)?,
1279 None => select::Language::Pointer,
1280 };
1281 select::place(language, &replacement.target, &mut payload, value).map_err(|reason| {
1282 ExecutionError::BadRequest {
1283 step: step.step_id.clone(),
1284 reason,
1285 }
1286 })?;
1287 }
1288 let content_type = request_body
1289 .content_type
1290 .clone()
1291 .unwrap_or_else(|| "application/json".to_owned());
1292 let bytes = match (&payload, content_type.contains("json")) {
1295 (Value::String(text), false) => text.clone().into_bytes(),
1296 _ => payload.to_string().into_bytes(),
1297 };
1298 Ok(Some(Body {
1299 bytes,
1300 value: payload,
1301 content_type,
1302 }))
1303}
1304
1305fn url(
1308 endpoint: &operation::Endpoint,
1309 path: &BTreeMap<String, Value>,
1310 query: &BTreeMap<String, Value>,
1311 querystring: Option<&str>,
1312) -> Result<String, String> {
1313 let mut filled = endpoint.path.clone();
1314 for (name, value) in path {
1315 filled = filled.replace(&format!("{{{name}}}"), &encode(&text(value)));
1316 }
1317 if let Some(start) = filled.find('{') {
1318 return Err(format!(
1319 "`{}` still has `{}` in it, which no parameter filled in",
1320 endpoint.path,
1321 &filled[start
1322 ..filled[start..]
1323 .find('}')
1324 .map_or(filled.len(), |end| start + end + 1)]
1325 ));
1326 }
1327 let mut url = format!("{}{filled}", endpoint.base);
1328 let pairs: Vec<String> = query
1329 .iter()
1330 .map(|(name, value)| format!("{}={}", encode(name), encode(&text(value))))
1331 .collect();
1332 let query = match (pairs.is_empty(), querystring) {
1333 (true, None) => String::new(),
1334 (true, Some(raw)) => raw.to_owned(),
1335 (false, None) => pairs.join("&"),
1336 (false, Some(raw)) => format!("{}&{raw}", pairs.join("&")),
1337 };
1338 if !query.is_empty() {
1339 url.push('?');
1340 url.push_str(&query);
1341 }
1342 url::Url::parse(&url).map_err(|error| format!("`{url}` is not a URL: {error}"))?;
1343 Ok(url)
1344}
1345
1346fn text(value: &Value) -> String {
1349 match value {
1350 Value::String(text) => text.clone(),
1351 other => other.to_string(),
1352 }
1353}
1354
1355fn encode(text: &str) -> String {
1357 let mut encoded = String::with_capacity(text.len());
1358 for byte in text.bytes() {
1359 match byte {
1360 b'A'..=b'Z' | b'a'..=b'z' | b'0'..=b'9' | b'-' | b'.' | b'_' | b'~' => {
1361 encoded.push(byte as char);
1362 }
1363 other => encoded.push_str(&format!("%{other:02X}")),
1364 }
1365 }
1366 encoded
1367}
1368
1369fn ordered_workflows<'d>(
1372 description: &'d Description,
1373 wanted: &'d Workflow,
1374) -> Result<Vec<&'d Workflow>, ExecutionError> {
1375 let mut ordered = Vec::new();
1376 let mut visiting = BTreeSet::new();
1377 let mut done = BTreeSet::new();
1378 visit(description, wanted, &mut ordered, &mut visiting, &mut done)?;
1379 Ok(ordered)
1380}
1381
1382fn visit<'d>(
1383 description: &'d Description,
1384 workflow: &'d Workflow,
1385 ordered: &mut Vec<&'d Workflow>,
1386 visiting: &mut BTreeSet<String>,
1387 done: &mut BTreeSet<String>,
1388) -> Result<(), ExecutionError> {
1389 if done.contains(&workflow.workflow_id) {
1390 return Ok(());
1391 }
1392 if !visiting.insert(workflow.workflow_id.clone()) {
1393 return Err(ExecutionError::Circular(workflow.workflow_id.clone()));
1394 }
1395 for id in &workflow.depends_on {
1396 let dependency = description
1397 .workflows
1398 .iter()
1399 .find(|candidate| &candidate.workflow_id == id)
1400 .ok_or_else(|| ExecutionError::UnknownWorkflow(id.clone()))?;
1401 visit(description, dependency, ordered, visiting, done)?;
1402 }
1403 visiting.remove(&workflow.workflow_id);
1404 done.insert(workflow.workflow_id.clone());
1405 ordered.push(workflow);
1406 Ok(())
1407}
1408
1409fn steps_named_by(step: &Step, description: &Description) -> BTreeSet<String> {
1423 let mut found = BTreeSet::new();
1424
1425 fn named_in(expression: &str) -> Option<String> {
1427 let rest = expression.strip_prefix("$steps.")?;
1428 let (id, _) = rest.split_once('.')?;
1429 Some(id.to_owned())
1430 }
1431 fn read(text: &str, found: &mut BTreeSet<String>) {
1432 found.extend(
1433 expression::references(text)
1434 .into_iter()
1435 .filter_map(named_in),
1436 );
1437 }
1438 fn read_condition(criterion: &Criterion, found: &mut BTreeSet<String>) {
1442 let simple = matches!(
1443 criterion.type_,
1444 None | Some(CriterionType::Simple(CriterionKind::Simple))
1445 );
1446 if simple {
1447 found.extend(
1448 criterion::expressions_in(&criterion.condition)
1449 .iter()
1450 .filter_map(|expression| named_in(expression)),
1451 );
1452 } else {
1453 found.extend(
1457 expression::interpolations(&criterion.condition)
1458 .into_iter()
1459 .filter_map(named_in),
1460 );
1461 }
1462 }
1463 fn read_value(value: &ValueOrSelector, found: &mut BTreeSet<String>) {
1464 match value {
1465 ValueOrSelector::Literal(literal) => read_literal(literal, found),
1466 ValueOrSelector::Selector(selector) => read(&selector.context, found),
1470 }
1471 }
1472 fn read_literal(value: &Value, found: &mut BTreeSet<String>) {
1473 match value {
1474 Value::String(text) => read(text, found),
1475 Value::Array(items) => items.iter().for_each(|item| read_literal(item, found)),
1476 Value::Object(members) => members
1477 .values()
1478 .for_each(|member| read_literal(member, found)),
1479 _ => {}
1480 }
1481 }
1482 fn read_parameters(
1483 list: &[ReusableOr<Parameter>],
1484 description: &Description,
1485 found: &mut BTreeSet<String>,
1486 ) {
1487 for entry in list {
1488 match entry {
1489 ReusableOr::Item(parameter) => read_value(¶meter.value, found),
1490 ReusableOr::Reusable(reusable) => {
1491 if let Some(overridden) = &reusable.value {
1495 read_literal(overridden, found);
1496 } else if let Some(parameter) = reusable
1497 .reference
1498 .strip_prefix("$components.parameters.")
1499 .and_then(|name| {
1500 description
1501 .components
1502 .as_ref()
1503 .and_then(|components| components.parameters.get(name))
1504 })
1505 {
1506 read_value(¶meter.value, found);
1507 }
1508 }
1509 }
1510 }
1511 }
1512 fn read_criteria(list: &[Criterion], found: &mut BTreeSet<String>) {
1513 for criterion in list {
1514 if let Some(context) = &criterion.context {
1515 read(context, found);
1516 }
1517 read_condition(criterion, found);
1518 }
1519 }
1520
1521 read_parameters(&step.parameters, description, &mut found);
1522 read_criteria(&step.success_criteria, &mut found);
1523 for output in step.outputs.values() {
1524 read_value(output, &mut found);
1525 }
1526 if let Some(body) = &step.request_body {
1527 if let Some(payload) = &body.payload {
1528 read_literal(payload, &mut found);
1529 }
1530 for replacement in &body.replacements {
1531 read_value(&replacement.value, &mut found);
1532 }
1533 }
1534 for entry in &step.on_success {
1535 if let Ok(action) = success_action(entry, description) {
1536 read_criteria(&action.criteria, &mut found);
1537 read_parameters(&action.parameters, description, &mut found);
1538 }
1539 }
1540 for entry in &step.on_failure {
1541 if let Ok(action) = failure_action(entry, description) {
1542 read_criteria(&action.criteria, &mut found);
1543 read_parameters(&action.parameters, description, &mut found);
1544 }
1545 }
1546
1547 found.remove(&step.step_id);
1548 found
1549}
1550
1551fn ordered_steps(
1555 workflow: &Workflow,
1556 description: &Description,
1557) -> Result<Vec<usize>, ExecutionError> {
1558 let index: BTreeMap<&str, usize> = workflow
1559 .steps
1560 .iter()
1561 .enumerate()
1562 .map(|(at, step)| (step.step_id.as_str(), at))
1563 .collect();
1564 let mut ordered = Vec::with_capacity(workflow.steps.len());
1565 let mut visiting = BTreeSet::new();
1566 let mut done = BTreeSet::new();
1567 for step in &workflow.steps {
1568 visit_step(
1569 workflow,
1570 description,
1571 &index,
1572 step,
1573 &mut ordered,
1574 &mut visiting,
1575 &mut done,
1576 )?;
1577 }
1578 Ok(ordered)
1579}
1580
1581fn visit_step(
1582 workflow: &Workflow,
1583 description: &Description,
1584 index: &BTreeMap<&str, usize>,
1585 step: &Step,
1586 ordered: &mut Vec<usize>,
1587 visiting: &mut BTreeSet<String>,
1588 done: &mut BTreeSet<String>,
1589) -> Result<(), ExecutionError> {
1590 if done.contains(&step.step_id) {
1591 return Ok(());
1592 }
1593 if !visiting.insert(step.step_id.clone()) {
1594 return Err(ExecutionError::Circular(step.step_id.clone()));
1595 }
1596 for id in &step.depends_on {
1597 let at = index
1598 .get(id.as_str())
1599 .ok_or_else(|| ExecutionError::UnknownStep {
1600 workflow: workflow.workflow_id.clone(),
1601 step: id.clone(),
1602 })?;
1603 visit_step(
1604 workflow,
1605 description,
1606 index,
1607 &workflow.steps[*at],
1608 ordered,
1609 visiting,
1610 done,
1611 )?;
1612 }
1613 for id in steps_named_by(step, description) {
1617 let Some(at) = index.get(id.as_str()) else {
1618 continue;
1619 };
1620 visit_step(
1621 workflow,
1622 description,
1623 index,
1624 &workflow.steps[*at],
1625 ordered,
1626 visiting,
1627 done,
1628 )?;
1629 }
1630 visiting.remove(&step.step_id);
1631 done.insert(step.step_id.clone());
1632 ordered.push(index[step.step_id.as_str()]);
1633 Ok(())
1634}