dex_sdk/sub_flow.rs
1// Copyright (c) 2026 Super Durable, Inc.
2//
3// Licensed under the Sustainable Use License 1.0.
4// You may not use this file except in compliance with the License.
5// See the LICENSE file in the repository root.
6//
7// SPDX-License-Identifier: LicenseRef-Sustainable-Use-1.0
8
9use std::any::TypeId;
10use std::time::Duration;
11
12use dex_protocol::dex::WaitingCondition;
13
14use crate::start_flow_options::InitialAttribute;
15use crate::wait::SubFlowDefinition;
16use crate::{
17 Attribute, AttributeMap, Condition, Context, Flow, FlowConfig, FlowResult,
18 FlowTimeoutHandlerOptions, FlowTimeoutPolicy, HandlerResult, RetryPolicy, SdkResult, Value,
19 value_mapper,
20};
21
22#[derive(Clone, Copy, Debug, Eq, PartialEq)]
23/// Controls how a generated SubFlow Flow ID resolves an existing execution.
24pub enum SubFlowReusePolicy {
25 /// Attaches to a running execution or returns its existing terminal result.
26 Attach,
27 /// Restarts abnormal executions, attaches while running, and returns completed results.
28 RestartIfPreviousExitsAbnormally,
29 /// Replaces any different existing execution, including a running one.
30 AlwaysRestart,
31}
32
33#[derive(Clone)]
34/// Configures one durable SubFlow Condition.
35///
36/// Changed in 1.25.0: the SubFlow always selects pull Workers. Other omitted
37/// configuration fields inherit the parent's effective settings. Dex generates its Flow ID and
38/// request ID. Builder methods preserve normal Flow start defaults when omitted.
39/// Matching generated request IDs always attach to an existing child, regardless of reuse policy.
40/// For different request IDs, the reuse policy decides whether to attach or restart.
41/// Completing an any-of wait does not stop non-winning SubFlows; use their generated Flow IDs
42/// with [`crate::Client::stop_flow`] when those children should stop.
43pub struct SubFlowOptions {
44 pub(crate) timeout: Option<Duration>,
45 pub(crate) timeout_policy: FlowTimeoutPolicy,
46 pub(crate) timeout_handler_options: Option<FlowTimeoutHandlerOptions>,
47 pub(crate) start_delay: Option<Duration>,
48 pub(crate) retry_policy: Option<RetryPolicy>,
49 pub(crate) config_override: Option<FlowConfig>,
50 pub(crate) attributes: Vec<InitialAttribute>,
51 pub(crate) reuse_policy: SubFlowReusePolicy,
52 pub(crate) condition_id: Option<String>,
53}
54
55impl SubFlowOptions {
56 /// Creates options using the abnormal-restart reuse policy and normal start defaults.
57 pub fn new() -> Self {
58 Self {
59 timeout: None,
60 timeout_policy: FlowTimeoutPolicy::Default,
61 timeout_handler_options: None,
62 start_delay: None,
63 retry_policy: None,
64 config_override: None,
65 attributes: Vec::new(),
66 reuse_policy: SubFlowReusePolicy::RestartIfPreviousExitsAbnormally,
67 condition_id: None,
68 }
69 }
70
71 /// Sets the maximum total SubFlow execution duration.
72 pub fn timeout(mut self, value: Duration) -> Self {
73 self.timeout = Some(value);
74 self
75 }
76
77 /// Selects what Dex does when the positive soft SubFlow timeout expires.
78 pub fn timeout_policy(mut self, value: FlowTimeoutPolicy) -> Self {
79 self.timeout_policy = value;
80 self
81 }
82
83 /// Configures execution and selective state loading for the target Flow's timeout handler.
84 pub fn timeout_handler_options(mut self, value: FlowTimeoutHandlerOptions) -> Self {
85 self.timeout_handler_options = Some(value);
86 self
87 }
88
89 /// Sets a delay before the SubFlow starting Step after server acceptance.
90 ///
91 /// The Server currently supports only a zero delay. It rejects a Step result that requests a
92 /// SubFlow with a positive delay as invalid input, so that Step attempt fails.
93 pub fn start_delay(mut self, value: Duration) -> Self {
94 self.start_delay = Some(value);
95 self
96 }
97
98 /// Sets a whole-Flow retry policy for the SubFlow.
99 ///
100 /// The Server does not support whole-Flow retry. It rejects a Step result that requests a
101 /// SubFlow with this option as invalid input, so that Step attempt fails. Configure
102 /// [`crate::RetryPolicy`] on the SubFlow's Step methods through [`crate::StepOptions`]
103 /// instead.
104 pub fn retry_policy(mut self, value: RetryPolicy) -> Self {
105 self.retry_policy = Some(value);
106 self
107 }
108
109 /// Applies fields over the inherited parent Flow configuration.
110 pub fn config_override(mut self, value: FlowConfig) -> Self {
111 self.config_override = Some(value);
112 self
113 }
114
115 /// Selects existing-execution resolution behavior.
116 pub fn reuse_policy(mut self, value: SubFlowReusePolicy) -> Self {
117 self.reuse_policy = value;
118 self
119 }
120
121 /// Assigns the stable ID required by [`crate::Wait::any_combination_of`].
122 pub fn condition_id(mut self, value: impl Into<String>) -> Self {
123 self.condition_id = Some(value.into());
124 self
125 }
126
127 /// Adds one initial singleton Attribute owned by the target SubFlow.
128 pub fn initial_attribute<T: Value>(mut self, attribute: &Attribute<T>, value: T) -> Self {
129 self.attributes
130 .push(InitialAttribute::for_attribute(attribute, value));
131 self
132 }
133
134 /// Adds one initial Attribute-map value for an instance.
135 /// Slash is prohibited in instance keys because it is a reserved character.
136 pub fn initial_attribute_map<T: Value>(
137 mut self,
138 attribute: &AttributeMap<T>,
139 instance: &str,
140 value: T,
141 ) -> Self {
142 self.attributes.push(InitialAttribute::for_attribute_map(
143 attribute, instance, value,
144 ));
145 self
146 }
147}
148
149impl Default for SubFlowOptions {
150 fn default() -> Self {
151 Self::new()
152 }
153}
154
155/// Creates durable SubFlow Conditions and reads their Execute results.
156///
157/// Construction encodes input and returns a Condition; Dex starts or attaches to the child when
158/// the Worker submits that wait. The target Flow must be in the Worker's Registry. The parent
159/// waits durably without holding a handler thread. Inspect [`FlowResult`] for child failure;
160/// a child's abnormal terminal status is a result rather than a local value-mapping error.
161/// Result indexes are zero-based within the SubFlow Conditions, not all Condition kinds.
162///
163/// Dex generates each child Flow ID as `SubFlow:{parent Flow ID}-{Step execution ID}-{index}`.
164/// The Step execution ID is `{Step type}-{n}`.
165/// Because these parts are joined with `-`, only a Step whose Step type contains no `-` can start
166/// SubFlows.
167/// When a Step whose Step type contains `-` returns a Wait with a SubFlow Condition, the Worker
168/// sends the result unchanged and logs a warning.
169/// The server rejects that result as an invalid Worker response.
170/// It fails the `wait_for` attempt without retry, so under the default policy the Flow fails.
171/// Rename the Step type, for example by overriding [`crate::Step::step_type`], to start
172/// SubFlows from it.
173///
174/// Changed in 1.45.0: a Step whose Step type contains `-` can no longer start SubFlows, because
175/// two different parents could otherwise generate the same child Flow ID.
176///
177/// ```no_run
178/// use dex_sdk::{Flow, SdkResult, SubFlow, SubFlowOptions, Wait};
179/// use std::time::Duration;
180/// fn wait_for_child<F: Flow>(flow: &F, input: F::StartInput) -> SdkResult<Wait> {
181/// let child = SubFlow::run_with_options(flow, input,
182/// SubFlowOptions::new().timeout(Duration::from_secs(60)))?;
183/// Ok(Wait::until(child))
184/// }
185/// ```
186pub struct SubFlow;
187
188impl SubFlow {
189 /// Creates a SubFlow Condition with default options.
190 ///
191 /// # Errors
192 ///
193 /// Returns a value-mapping error when `input` cannot be encoded.
194 /// The server rejects the SubFlow and fails the `wait_for` attempt without retry when the
195 /// current Step type contains `-`. See [`SubFlow`] for this rule.
196 ///
197 /// Changed in 1.45.0: a Step whose Step type contains `-` can no longer start SubFlows.
198 pub fn run<SomeFlow: Flow>(
199 flow: &SomeFlow,
200 input: SomeFlow::StartInput,
201 ) -> SdkResult<Condition> {
202 Self::run_with_options(flow, input, SubFlowOptions::new())
203 }
204
205 /// Creates a SubFlow Condition with explicit options.
206 ///
207 /// # Errors
208 ///
209 /// Returns a value-mapping error when `input` cannot be encoded. Worker mapping later validates
210 /// the exact registered Rust Flow type, starting Step, Attributes, and option durations.
211 /// The server rejects the SubFlow and fails the `wait_for` attempt without retry when the
212 /// current Step type contains `-`. See [`SubFlow`] for this rule.
213 ///
214 /// Changed in 1.45.0: a Step whose Step type contains `-` can no longer start SubFlows.
215 pub fn run_with_options<SomeFlow: Flow>(
216 flow: &SomeFlow,
217 input: SomeFlow::StartInput,
218 options: SubFlowOptions,
219 ) -> SdkResult<Condition> {
220 Ok(Condition::sub_flow(SubFlowDefinition {
221 flow_type: flow.flow_type(),
222 type_id: TypeId::of::<SomeFlow>(),
223 input: value_mapper::encode(&input)?,
224 options,
225 }))
226 }
227
228 /// Returns the first SubFlow result during Step `execute`.
229 ///
230 /// # Errors
231 ///
232 /// Returns [`crate::HandlerError`] outside Execute or when the selected SubFlow result is
233 /// unavailable or malformed. See [`SubFlow`] for constructing the matching wait.
234 pub fn condition_result(context: &Context) -> HandlerResult<FlowResult> {
235 Self::condition_result_at(context, 0)
236 }
237
238 /// Returns one stable-indexed SubFlow result during Step `execute`.
239 ///
240 /// # Errors
241 ///
242 /// Returns [`crate::HandlerError`] outside Execute or when the selected SubFlow result is
243 /// unavailable or malformed. See [`SubFlow`] for constructing the matching wait.
244 pub fn condition_result_at(context: &Context, index: usize) -> HandlerResult<FlowResult> {
245 context.sub_flow_result(index)
246 }
247
248 /// Returns the generated Flow ID for the first SubFlow Condition during Step `execute`.
249 ///
250 /// # Errors
251 ///
252 /// Returns [`crate::HandlerError`] outside Execute or when the selected SubFlow result is
253 /// unavailable or malformed. See [`SubFlow`] for constructing the matching wait.
254 pub fn flow_id(context: &Context) -> HandlerResult<String> {
255 Self::flow_id_at(context, 0)
256 }
257
258 /// Returns the generated Flow ID for one stable-indexed SubFlow Condition during `execute`.
259 ///
260 /// # Errors
261 ///
262 /// Returns [`crate::HandlerError`] outside Execute or when the selected SubFlow result is
263 /// unavailable or malformed. See [`SubFlow`] for constructing the matching wait.
264 pub fn flow_id_at(context: &Context, index: usize) -> HandlerResult<String> {
265 context.sub_flow_id(index)
266 }
267}
268
269/// Warns when a dashed Step type starts SubFlows; the server rejects them without retry.
270pub(crate) fn warn_dashed_sub_flow_step_type(
271 step_type: &str,
272 condition: Option<WaitingCondition>,
273) -> Option<WaitingCondition> {
274 let has_sub_flows = condition
275 .as_ref()
276 .is_some_and(|condition| !condition.sub_flow_conditions.is_empty());
277 if has_sub_flows && step_type.contains('-') {
278 tracing::warn!(
279 step_type,
280 "a Step type that contains '-' cannot start SubFlows; the server rejects this result \
281 without retry; rename the Step type"
282 );
283 }
284 condition
285}
286
287#[cfg(test)]
288#[path = "sub_flow_tests.rs"]
289mod tests;