Internal: execute workflow steps starting from `start_index`, without acquiring the semaphore. Called by both [`execute_run`] and [`execute_from_step`] after they have already acquired a permit. On error, returns `(WorkflowError, PartialProgress)` so callers can persist the trace of steps completed before the failure.
(
engine: &WorkflowEngine,
community_id: CommunityId,
run_id: Uuid,
def: &WorkflowDef,
trigger_ctx: &TriggerContext,
start_index: usize,
initial_outputs: Option<HashMap<Str
| 1092 | /// On error, returns `(WorkflowError, PartialProgress)` so callers can persist |
| 1093 | /// the trace of steps completed before the failure. |
| 1094 | async fn execute_steps( |
| 1095 | engine: &WorkflowEngine, |
| 1096 | community_id: CommunityId, |
| 1097 | run_id: Uuid, |
| 1098 | def: &WorkflowDef, |
| 1099 | trigger_ctx: &TriggerContext, |
| 1100 | start_index: usize, |
| 1101 | initial_outputs: Option<HashMap<String, JsonValue>>, |
| 1102 | ) -> Result<ExecutionResult, (WorkflowError, crate::error::PartialProgress)> { |
| 1103 | let mut step_outputs: HashMap<String, JsonValue> = initial_outputs.unwrap_or_default(); |
| 1104 | let mut trace: Vec<JsonValue> = Vec::new(); |
| 1105 | |
| 1106 | for (i, step) in def.steps.iter().enumerate() { |
| 1107 | if i < start_index { |
| 1108 | debug!(run_id = %run_id, step = %step.id, "Skipping already-executed step"); |
| 1109 | continue; |
| 1110 | } |
| 1111 | |
| 1112 | if let Some(expr) = &step.if_expr { |
| 1113 | match evaluate_condition(expr, trigger_ctx, &step_outputs).await { |
| 1114 | Ok(true) => { |
| 1115 | debug!(run_id = %run_id, step = %step.id, "Condition true — running step"); |
| 1116 | } |
| 1117 | Ok(false) => { |
| 1118 | info!(run_id = %run_id, step = %step.id, "Condition false — skipping step"); |
| 1119 | trace.push(serde_json::json!({ |
| 1120 | "step_id": step.id, |
| 1121 | "status": "skipped", |
| 1122 | })); |
| 1123 | continue; |
| 1124 | } |
| 1125 | Err(e) => { |
| 1126 | warn!(run_id = %run_id, step = %step.id, "Condition error: {e}"); |
| 1127 | let progress = crate::error::PartialProgress { |
| 1128 | step_index: i, |
| 1129 | trace, |
| 1130 | }; |
| 1131 | return Err((e, progress)); |
| 1132 | } |
| 1133 | } |
| 1134 | } |
| 1135 | |
| 1136 | let resolved_action = match resolve_step_templates(step, trigger_ctx, &step_outputs) { |
| 1137 | Ok(a) => a, |
| 1138 | Err(e) => { |
| 1139 | let progress = crate::error::PartialProgress { |
| 1140 | step_index: i, |
| 1141 | trace, |
| 1142 | }; |
| 1143 | return Err((e, progress)); |
| 1144 | } |
| 1145 | }; |
| 1146 | |
| 1147 | let timeout_secs = step |
| 1148 | .timeout_secs |
| 1149 | .unwrap_or(engine.config.default_timeout_secs); |
| 1150 | let dispatch_result = tokio::time::timeout( |
| 1151 | std::time::Duration::from_secs(timeout_secs), |
no test coverage detected