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::{strip_continued_suffix, timestamp_now, WorkflowCtx};
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(
80        &self,
81        parent_id: &str,
82    ) -> Result<Vec<WorkflowRecord>> {
83        self.store.list_child_workflows(parent_id).await
84    }
85
86    pub async fn continue_as_new(
87        &self,
88        workflow_id: &str,
89        input: Option<&str>,
90        explicit_new_id: Option<&str>,
91    ) -> Result<WorkflowRecord> {
92        let old_wf = self
93            .store
94            .get_workflow(workflow_id)
95            .await?
96            .ok_or_else(|| anyhow::anyhow!("workflow not found: {workflow_id}"))?;
97
98        let old_status = WorkflowStatus::from_str(&old_wf.status)
99            .map_err(|e| anyhow::anyhow!(e))?;
100
101        // Only close the old run when it's still in-flight. A workflow
102        // that's already CANCELLED / FAILED / COMPLETED / TERMINATED
103        // stays that way — overwriting the terminal status would lose
104        // audit history (e.g. a cancelled run flipping to COMPLETED
105        // when the operator hits "Start a fresh run" on the row).
106        if !old_status.is_terminal() {
107            self.store
108                .update_workflow_status(workflow_id, WorkflowStatus::Completed, None, None)
109                .await?;
110        }
111
112        // Start a new run with the same type, namespace, and queue.
113        // Naming:
114        //   1. Caller-provided explicit id (e.g. dashboard combobox) —
115        //      honoured verbatim.
116        //   2. Otherwise derive: strip any existing `-continued-<digits>`
117        //      suffix from the source so sequential continues don't
118        //      stack (`demo-1` → `demo-1-continued-1` → `demo-1-continued-2`
119        //      instead of the compounding `demo-1-continued-1-continued-2`).
120        let new_id = match explicit_new_id {
121            Some(id) => id.to_string(),
122            None => {
123                let base = strip_continued_suffix(workflow_id);
124                format!("{base}-continued-{}", timestamp_now() as u64)
125            }
126        };
127        self.start_workflow(
128            &old_wf.namespace,
129            &old_wf.workflow_type,
130            &new_id,
131            input,
132            &old_wf.task_queue,
133            old_wf.search_attributes.as_deref(),
134        )
135        .await
136    }
137}
assay_workflow/children.rs

assay_workflow/
children.rs
