dex_sdk/
start_flow_options.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::time::Duration;
10
11use std::sync::Arc;
12
13use dex_protocol::dex::{AttributeSyncConfig, IndexConfig};
14
15use crate::registry::physical_name;
16use crate::step::{ErasedValue, TypedValue};
17use crate::{Attribute, AttributeMap, FlowConfig, FlowTimeoutHandlerOptions, RetryPolicy, Value};
18
19#[derive(Clone, Copy, Debug, Eq, PartialEq)]
20/// Controls whether [`crate::Client::start_flow_with_options`] may reuse a Flow ID.
21pub enum IdReusePolicy {
22    /// Uses the Dex server's configured reuse policy.
23    Default,
24    /// Allows reuse only when the previous run failed.
25    AllowIfPreviousFailed,
26    /// Allows reuse when no run with the ID is currently active.
27    AllowIfNotRunning,
28    /// Rejects reuse regardless of the previous run status.
29    Disallow,
30    /// Terminates an active run before starting the replacement.
31    TerminateIfRunning,
32}
33
34#[derive(Clone, Copy, Debug, Eq, PartialEq)]
35/// Controls how Dex responds when a positive soft Flow timeout expires.
36pub enum FlowTimeoutPolicy {
37    /// Uses the Flow's registered timeout handler when present, and fails otherwise.
38    Default,
39    /// Fails with [`crate::FlowErrorType::FlowTimeout`] and permits Flow retries.
40    Fail,
41    /// Cancels without retrying the Flow.
42    Cancel,
43    /// Invokes the registered [`crate::FlowTimeoutHandler`] as one retryable logical execution.
44    Handler,
45}
46
47#[derive(Clone)]
48/// Configures a new Flow execution.
49///
50/// All optional fields are absent by default. Initial Attribute values are encoded when the start
51/// request is sent. Use a stable request ID to make retries of the same logical request idempotent.
52///
53/// # Examples
54///
55/// ```
56/// use dex_sdk::{Attribute, IdReusePolicy, StartFlowOptions};
57/// use std::time::Duration;
58///
59/// let customer = Attribute::<String>::new("customer");
60/// let options = StartFlowOptions::new()
61///     .timeout(Duration::from_secs(3_600))
62///     .id_reuse_policy(IdReusePolicy::AllowIfNotRunning)
63///     .initial_attribute(&customer, "customer-42".to_owned())
64///     .request_id("create-order-42");
65/// ```
66pub struct StartFlowOptions {
67    pub(crate) timeout: Option<Duration>,
68    pub(crate) timeout_policy: FlowTimeoutPolicy,
69    pub(crate) timeout_handler_options: Option<FlowTimeoutHandlerOptions>,
70    pub(crate) start_delay: Option<Duration>,
71    pub(crate) id_reuse_policy: IdReusePolicy,
72    pub(crate) retry_policy: Option<RetryPolicy>,
73    pub(crate) config_override: Option<FlowConfig>,
74    pub(crate) ignore_already_started: bool,
75    pub(crate) request_id: Option<String>,
76    pub(crate) attributes: Vec<InitialAttribute>,
77}
78
79#[derive(Clone)]
80pub(crate) struct InitialAttribute {
81    pub(crate) key: String,
82    pub(crate) value: Arc<dyn ErasedValue>,
83    pub(crate) index_config: Option<IndexConfig>,
84    pub(crate) sync_config: Option<AttributeSyncConfig>,
85}
86
87impl InitialAttribute {
88    pub(crate) fn for_attribute<T: Value>(attribute: &Attribute<T>, value: T) -> Self {
89        Self {
90            key: attribute.name().to_string(),
91            value: Arc::new(TypedValue(value)),
92            index_config: attribute.index().map(|index| index.proto_config(false)),
93            sync_config: attribute.sync_config(),
94        }
95    }
96
97    pub(crate) fn for_attribute_map<T: Value>(
98        attribute: &AttributeMap<T>,
99        instance: &str,
100        value: T,
101    ) -> Self {
102        Self {
103            key: physical_name(attribute.name(), instance),
104            value: Arc::new(TypedValue(value)),
105            index_config: attribute.index().map(|index| index.proto_config(true)),
106            sync_config: attribute.sync_config(),
107        }
108    }
109}
110
111impl StartFlowOptions {
112    /// Creates start options that preserve all server defaults.
113    pub fn new() -> Self {
114        Self {
115            timeout: None,
116            timeout_policy: FlowTimeoutPolicy::Default,
117            timeout_handler_options: None,
118            start_delay: None,
119            id_reuse_policy: IdReusePolicy::Default,
120            retry_policy: None,
121            config_override: None,
122            ignore_already_started: false,
123            request_id: None,
124            attributes: Vec::new(),
125        }
126    }
127
128    /// Sets Dex's durable soft timeout. A zero duration disables it.
129    pub fn timeout(mut self, value: Duration) -> Self {
130        self.timeout = Some(value);
131        self
132    }
133
134    /// Sets the action taken when the positive Flow timeout expires.
135    ///
136    /// Handler is rejected before start when the registered Flow has no timeout handler.
137    pub fn timeout_policy(mut self, value: FlowTimeoutPolicy) -> Self {
138        self.timeout_policy = value;
139        self
140    }
141
142    /// Configures execution and selective state loading for the Flow timeout handler.
143    ///
144    /// Dex rejects this option unless the Flow has a positive timeout and resolves to
145    /// [`FlowTimeoutPolicy::Handler`].
146    pub fn timeout_handler_options(mut self, value: FlowTimeoutHandlerOptions) -> Self {
147        self.timeout_handler_options = Some(value);
148        self
149    }
150
151    /// Delays the first Step after the server accepts the Flow.
152    pub fn start_delay(mut self, value: Duration) -> Self {
153        self.start_delay = Some(value);
154        self
155    }
156
157    /// Sets how the server handles an existing Flow with the same ID.
158    pub fn id_reuse_policy(mut self, value: IdReusePolicy) -> Self {
159        self.id_reuse_policy = value;
160        self
161    }
162
163    /// Sets the whole-Flow retry policy after a terminal failure.
164    pub fn retry_policy(mut self, value: RetryPolicy) -> Self {
165        self.retry_policy = Some(value);
166        self
167    }
168
169    /// Adds an initial value for a declared Attribute.
170    pub fn initial_attribute<T: Value>(mut self, attribute: &Attribute<T>, value: T) -> Self {
171        self.attributes
172            .push(InitialAttribute::for_attribute(attribute, value));
173        self
174    }
175
176    /// Adds an initial value for one Attribute-map instance.
177    /// Slash is prohibited in instance keys because it is a reserved character.
178    pub fn initial_attribute_map<T: Value>(
179        mut self,
180        attribute: &AttributeMap<T>,
181        instance: &str,
182        value: T,
183    ) -> Self {
184        self.attributes.push(InitialAttribute::for_attribute_map(
185            attribute, instance, value,
186        ));
187        self
188    }
189
190    /// Overrides the registered Flow configuration for this execution.
191    pub fn config_override(mut self, value: FlowConfig) -> Self {
192        self.config_override = Some(value);
193        self
194    }
195
196    /// Returns the existing run ID instead of an error when the Flow already exists.
197    pub fn ignore_already_started(mut self, value: bool) -> Self {
198        self.ignore_already_started = value;
199        self
200    }
201
202    /// Sets the idempotency key for this start request.
203    pub fn request_id(mut self, value: impl Into<String>) -> Self {
204        self.request_id = Some(value.into());
205        self
206    }
207}
208
209impl Default for StartFlowOptions {
210    fn default() -> Self {
211        Self::new()
212    }
213}
214
215#[cfg(test)]
216mod tests {
217    use super::StartFlowOptions;
218    use crate::{Attribute, AttributeMap};
219
220    #[test]
221    fn initial_attributes_retain_sync_configuration() {
222        let plain = Attribute::<String>::new("plain");
223        let synced = Attribute::<String>::new("synced").sync_to_attribute_store();
224        let synced_map = AttributeMap::<String>::new("map").sync_to_attribute_store();
225        let options = StartFlowOptions::new()
226            .initial_attribute(&plain, "plain".to_string())
227            .initial_attribute(&synced, "synced".to_string())
228            .initial_attribute_map(&synced_map, "tenant-1", "mapped".to_string());
229
230        assert!(options.attributes[0].sync_config.is_none());
231        assert_eq!(
232            options.attributes[1]
233                .sync_config
234                .as_ref()
235                .map(|config| config.enabled),
236            Some(true)
237        );
238        assert_eq!(
239            options.attributes[2]
240                .sync_config
241                .as_ref()
242                .map(|config| config.enabled),
243            Some(true)
244        );
245    }
246}
dex_sdk/start_flow_options.rs

dex_sdk/
start_flow_options.rs
