beam_core/workflow_runtime/
loop.rs

1use anyhow::Result;
2
3use crate::workflow_orchestrator::OrchestratorAction;
4use crate::{EventDraft, EventLog, WorkflowActor};
5
6pub async fn start_loop(log: &mut EventLog, action: &crate::OrchestratorAction) -> Result<()> {
7    if let OrchestratorAction::StartLoop {
8        node_id,
9        max_iterations,
10    } = action
11    {
12        let _ = log.append(EventDraft {
13            event_type: "loopStarted".to_string(),
14            actor: WorkflowActor::Scheduler,
15            payload: serde_json::json!({
16                "loopId": node_id,
17                "maxIterations": max_iterations,
18            }),
19            timestamp: None,
20            payload_hash: None,
21        })?;
22        Ok(())
23    } else {
24        anyhow::bail!("start_loop called with wrong action")
25    }
26}
27
28pub async fn start_loop_iteration(
29    log: &mut EventLog,
30    action: &crate::OrchestratorAction,
31) -> Result<()> {
32    if let OrchestratorAction::StartLoopIteration { node_id, iteration } = action {
33        let _ = log.append(EventDraft {
34            event_type: "loopIterationStarted".to_string(),
35            actor: WorkflowActor::Scheduler,
36            payload: serde_json::json!({
37                "loopId": node_id,
38                "iteration": iteration,
39            }),
40            timestamp: None,
41            payload_hash: None,
42        })?;
43        Ok(())
44    } else {
45        anyhow::bail!("start_loop_iteration called with wrong action")
46    }
47}
48
49pub async fn finish_loop_iteration(
50    log: &mut EventLog,
51    action: &crate::OrchestratorAction,
52) -> Result<()> {
53    if let OrchestratorAction::FinishLoopIteration {
54        node_id,
55        iteration,
56        resolution,
57        decision_activity_id,
58        wait_resolved_event_id,
59        by,
60        comment,
61        timed_out,
62    } = action
63    {
64        let _ = log.append(EventDraft {
65            event_type: "loopIterationFinished".to_string(),
66            actor: WorkflowActor::Scheduler,
67            payload: serde_json::json!({
68                "loopId": node_id,
69                "iteration": iteration,
70                "resolution": resolution,
71                "decisionActivityId": decision_activity_id,
72                "waitResolvedEventId": wait_resolved_event_id,
73                "by": by,
74                "comment": comment,
75                "timedOut": timed_out,
76            }),
77            timestamp: None,
78            payload_hash: None,
79        })?;
80        Ok(())
81    } else {
82        anyhow::bail!("finish_loop_iteration called with wrong action")
83    }
84}
85
86pub async fn finish_loop(log: &mut EventLog, action: &crate::OrchestratorAction) -> Result<()> {
87    if let OrchestratorAction::FinishLoop {
88        node_id,
89        final_iteration,
90        resolution,
91        output_ref,
92        error_code,
93        error_class,
94    } = action
95    {
96        let _ = log.append(EventDraft {
97            event_type: "loopFinished".to_string(),
98            actor: WorkflowActor::Scheduler,
99            payload: serde_json::json!({
100                "loopId": node_id,
101                "finalIteration": final_iteration,
102                "resolution": resolution,
103                "outputRef": output_ref,
104                "errorCode": error_code,
105                "errorClass": error_class,
106            }),
107            timestamp: None,
108            payload_hash: None,
109        })?;
110        Ok(())
111    } else {
112        anyhow::bail!("finish_loop called with wrong action")
113    }
114}
beam_core/workflow_runtime/loop.rs

beam_core/workflow_runtime/
loop.rs
