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 crate::start_flow_options::InitialAttribute;
13use crate::wait::SubFlowDefinition;
14use crate::{
15    Attribute, AttributeMap, Condition, Context, Flow, FlowConfig, FlowResult,
16    FlowTimeoutHandlerOptions, FlowTimeoutPolicy, HandlerResult, RetryPolicy, SdkResult, Value,
17    value_mapper,
18};
19
20#[derive(Clone, Copy, Debug, Eq, PartialEq)]
21/// Controls how a generated SubFlow Flow ID resolves an existing execution.
22pub enum SubFlowReusePolicy {
23    /// Attaches to a running execution or returns its existing terminal result.
24    Attach,
25    /// Restarts abnormal executions, attaches while running, and returns completed results.
26    RestartIfPreviousExitsAbnormally,
27    /// Replaces any different existing execution, including a running one.
28    AlwaysRestart,
29}
30
31#[derive(Clone)]
32/// Configures one durable SubFlow Condition.
33///
34/// The SubFlow always selects pull Workers, overriding any parent push target. Other omitted
35/// configuration fields inherit the parent's effective settings. Dex generates its Flow ID and
36/// request ID. Builder methods preserve normal Flow start defaults when omitted.
37pub struct SubFlowOptions {
38    pub(crate) timeout: Option<Duration>,
39    pub(crate) timeout_policy: FlowTimeoutPolicy,
40    pub(crate) timeout_handler_options: Option<FlowTimeoutHandlerOptions>,
41    pub(crate) start_delay: Option<Duration>,
42    pub(crate) retry_policy: Option<RetryPolicy>,
43    pub(crate) config_override: Option<FlowConfig>,
44    pub(crate) attributes: Vec<InitialAttribute>,
45    pub(crate) reuse_policy: SubFlowReusePolicy,
46    pub(crate) condition_id: Option<String>,
47}
48
49impl SubFlowOptions {
50    /// Creates options using the abnormal-restart reuse policy and normal start defaults.
51    pub fn new() -> Self {
52        Self {
53            timeout: None,
54            timeout_policy: FlowTimeoutPolicy::Default,
55            timeout_handler_options: None,
56            start_delay: None,
57            retry_policy: None,
58            config_override: None,
59            attributes: Vec::new(),
60            reuse_policy: SubFlowReusePolicy::RestartIfPreviousExitsAbnormally,
61            condition_id: None,
62        }
63    }
64
65    /// Sets the maximum total SubFlow execution duration.
66    pub fn timeout(mut self, value: Duration) -> Self {
67        self.timeout = Some(value);
68        self
69    }
70
71    /// Selects what Dex does when the positive soft SubFlow timeout expires.
72    pub fn timeout_policy(mut self, value: FlowTimeoutPolicy) -> Self {
73        self.timeout_policy = value;
74        self
75    }
76
77    /// Configures execution and selective state loading for the target Flow's timeout handler.
78    pub fn timeout_handler_options(mut self, value: FlowTimeoutHandlerOptions) -> Self {
79        self.timeout_handler_options = Some(value);
80        self
81    }
82
83    /// Delays the SubFlow starting Step after server acceptance.
84    pub fn start_delay(mut self, value: Duration) -> Self {
85        self.start_delay = Some(value);
86        self
87    }
88
89    /// Sets whole-Flow retry behavior after abnormal completion.
90    pub fn retry_policy(mut self, value: RetryPolicy) -> Self {
91        self.retry_policy = Some(value);
92        self
93    }
94
95    /// Applies fields over the inherited parent Flow configuration.
96    pub fn config_override(mut self, value: FlowConfig) -> Self {
97        self.config_override = Some(value);
98        self
99    }
100
101    /// Selects existing-execution resolution behavior.
102    pub fn reuse_policy(mut self, value: SubFlowReusePolicy) -> Self {
103        self.reuse_policy = value;
104        self
105    }
106
107    /// Assigns the stable ID required by [`crate::Wait::any_combination_of`].
108    pub fn condition_id(mut self, value: impl Into<String>) -> Self {
109        self.condition_id = Some(value.into());
110        self
111    }
112
113    /// Adds one initial singleton Attribute owned by the target SubFlow.
114    pub fn initial_attribute<T: Value>(mut self, attribute: &Attribute<T>, value: T) -> Self {
115        self.attributes
116            .push(InitialAttribute::for_attribute(attribute, value));
117        self
118    }
119
120    /// Adds one initial Attribute-map value for an instance.
121    /// Slash is prohibited in instance keys because it is a reserved character.
122    pub fn initial_attribute_map<T: Value>(
123        mut self,
124        attribute: &AttributeMap<T>,
125        instance: &str,
126        value: T,
127    ) -> Self {
128        self.attributes.push(InitialAttribute::for_attribute_map(
129            attribute, instance, value,
130        ));
131        self
132    }
133}
134
135impl Default for SubFlowOptions {
136    fn default() -> Self {
137        Self::new()
138    }
139}
140
141/// Creates durable SubFlow Conditions and reads their Execute results.
142pub struct SubFlow;
143
144impl SubFlow {
145    /// Creates a SubFlow Condition with default options.
146    ///
147    /// # Errors
148    ///
149    /// Returns a value-mapping error when `input` cannot be encoded.
150    pub fn run<SomeFlow: Flow>(
151        flow: &SomeFlow,
152        input: SomeFlow::StartInput,
153    ) -> SdkResult<Condition> {
154        Self::run_with_options(flow, input, SubFlowOptions::new())
155    }
156
157    /// Creates a SubFlow Condition with explicit options.
158    ///
159    /// # Errors
160    ///
161    /// Returns a value-mapping error when `input` cannot be encoded. Worker mapping later validates
162    /// the exact registered Rust Flow type, starting Step, Attributes, and option durations.
163    pub fn run_with_options<SomeFlow: Flow>(
164        flow: &SomeFlow,
165        input: SomeFlow::StartInput,
166        options: SubFlowOptions,
167    ) -> SdkResult<Condition> {
168        Ok(Condition::sub_flow(SubFlowDefinition {
169            flow_type: flow.flow_type(),
170            type_id: TypeId::of::<SomeFlow>(),
171            input: value_mapper::encode(&input)?,
172            options,
173        }))
174    }
175
176    /// Returns the first SubFlow result during Step `execute`.
177    pub fn condition_result(context: &Context) -> HandlerResult<FlowResult> {
178        Self::condition_result_at(context, 0)
179    }
180
181    /// Returns one stable-indexed SubFlow result during Step `execute`.
182    pub fn condition_result_at(context: &Context, index: usize) -> HandlerResult<FlowResult> {
183        context.sub_flow_result(index)
184    }
185
186    /// Returns the generated Flow ID for the first SubFlow Condition during Step `execute`.
187    pub fn flow_id(context: &Context) -> HandlerResult<String> {
188        Self::flow_id_at(context, 0)
189    }
190
191    /// Returns the generated Flow ID for one stable-indexed SubFlow Condition during `execute`.
192    pub fn flow_id_at(context: &Context, index: usize) -> HandlerResult<String> {
193        context.sub_flow_id(index)
194    }
195}
dex_sdk/sub_flow.rs

dex_sdk/
sub_flow.rs
