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