dex_sdk/
sub_flow.rs

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

dex_sdk/
sub_flow.rs
