assay_workflow/
children.rs

1//! Child workflow operations and continue-as-new.
2
3use std::str::FromStr;
4
5use anyhow::Result;
6
7use crate::ctx::{WorkflowCtx, strip_continued_suffix, timestamp_now};
8use crate::store::WorkflowStore;
9use crate::types::*;
10
11impl<S: WorkflowStore> WorkflowCtx<S> {
12    pub async fn start_child_workflow(
13        &self,
14        namespace: &str,
15        parent_id: &str,
16        workflow_type: &str,
17        workflow_id: &str,
18        input: Option<&str>,
19        task_queue: &str,
20    ) -> Result<WorkflowRecord> {
21        let now = timestamp_now();
22        let run_id = format!("run-{workflow_id}-{}", now as u64);
23
24        let wf = WorkflowRecord {
25            id: workflow_id.to_string(),
26            namespace: namespace.to_string(),
27            run_id,
28            workflow_type: workflow_type.to_string(),
29            task_queue: task_queue.to_string(),
30            status: "PENDING".to_string(),
31            input: input.map(String::from),
32            result: None,
33            error: None,
34            parent_id: Some(parent_id.to_string()),
35            claimed_by: None,
36            search_attributes: None,
37            archived_at: None,
38            archive_uri: None,
39            created_at: now,
40            updated_at: now,
41            completed_at: None,
42        };
43
44        self.store.create_workflow(&wf).await?;
45
46        // Record events on both parent and child
47        self.store
48            .append_event(&WorkflowEvent {
49                id: None,
50                workflow_id: workflow_id.to_string(),
51                seq: 1,
52                event_type: "WorkflowStarted".to_string(),
53                payload: input.map(String::from),
54                timestamp: now,
55            })
56            .await?;
57
58        let parent_seq = self.store.get_event_count(parent_id).await? as i32 + 1;
59        self.store
60            .append_event(&WorkflowEvent {
61                id: None,
62                workflow_id: parent_id.to_string(),
63                seq: parent_seq,
64                event_type: "ChildWorkflowStarted".to_string(),
65                payload: Some(
66                    serde_json::json!({
67                        "child_workflow_id": workflow_id,
68                        "workflow_type": workflow_type,
69                    })
70                    .to_string(),
71                ),
72                timestamp: now,
73            })
74            .await?;
75
76        Ok(wf)
77    }
78
79    pub async fn list_child_workflows(&self, parent_id: &str) -> Result<Vec<WorkflowRecord>> {
80        self.store.list_child_workflows(parent_id).await
81    }
82
83    pub async fn continue_as_new(
84        &self,
85        workflow_id: &str,
86        input: Option<&str>,
87        explicit_new_id: Option<&str>,
88    ) -> Result<WorkflowRecord> {
89        let old_wf = self
90            .store
91            .get_workflow(workflow_id)
92            .await?
93            .ok_or_else(|| anyhow::anyhow!("workflow not found: {workflow_id}"))?;
94
95        let old_status =
96            WorkflowStatus::from_str(&old_wf.status).map_err(|e| anyhow::anyhow!(e))?;
97
98        // Only close the old run when it's still in-flight. A workflow
99        // that's already CANCELLED / FAILED / COMPLETED / TERMINATED
100        // stays that way — overwriting the terminal status would lose
101        // audit history (e.g. a cancelled run flipping to COMPLETED
102        // when the operator hits "Start a fresh run" on the row).
103        if !old_status.is_terminal() {
104            self.store
105                .update_workflow_status(workflow_id, WorkflowStatus::Completed, None, None)
106                .await?;
107        }
108
109        // Start a new run with the same type, namespace, and queue.
110        // Naming:
111        //   1. Caller-provided explicit id (e.g. dashboard combobox) —
112        //      honoured verbatim.
113        //   2. Otherwise derive: strip any existing `-continued-<digits>`
114        //      suffix from the source so sequential continues don't
115        //      stack (`demo-1` → `demo-1-continued-1` → `demo-1-continued-2`
116        //      instead of the compounding `demo-1-continued-1-continued-2`).
117        let new_id = match explicit_new_id {
118            Some(id) => id.to_string(),
119            None => {
120                let base = strip_continued_suffix(workflow_id);
121                format!("{base}-continued-{}", timestamp_now() as u64)
122            }
123        };
124        self.start_workflow(
125            &old_wf.namespace,
126            &old_wf.workflow_type,
127            &new_id,
128            input,
129            &old_wf.task_queue,
130            old_wf.search_attributes.as_deref(),
131        )
132        .await
133    }
134}
assay_workflow/children.rs

assay_workflow/
children.rs
