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, 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`] once.
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) start_delay: Option<Duration>,
70    pub(crate) id_reuse_policy: IdReusePolicy,
71    pub(crate) retry_policy: Option<RetryPolicy>,
72    pub(crate) config_override: Option<FlowConfig>,
73    pub(crate) ignore_already_started: bool,
74    pub(crate) request_id: Option<String>,
75    pub(crate) attributes: Vec<InitialAttribute>,
76}
77
78#[derive(Clone)]
79pub(crate) struct InitialAttribute {
80    pub(crate) key: String,
81    pub(crate) value: Arc<dyn ErasedValue>,
82    pub(crate) index_config: Option<IndexConfig>,
83    pub(crate) sync_config: Option<AttributeSyncConfig>,
84}
85
86impl InitialAttribute {
87    pub(crate) fn for_attribute<T: Value>(attribute: &Attribute<T>, value: T) -> Self {
88        Self {
89            key: attribute.name().to_string(),
90            value: Arc::new(TypedValue(value)),
91            index_config: attribute.index().map(|index| index.proto_config(false)),
92            sync_config: attribute.sync_config(),
93        }
94    }
95
96    pub(crate) fn for_attribute_map<T: Value>(
97        attribute: &AttributeMap<T>,
98        instance: &str,
99        value: T,
100    ) -> Self {
101        Self {
102            key: physical_name(attribute.name(), instance),
103            value: Arc::new(TypedValue(value)),
104            index_config: attribute.index().map(|index| index.proto_config(true)),
105            sync_config: attribute.sync_config(),
106        }
107    }
108}
109
110impl StartFlowOptions {
111    /// Creates start options that preserve all server defaults.
112    pub fn new() -> Self {
113        Self {
114            timeout: None,
115            timeout_policy: FlowTimeoutPolicy::Default,
116            start_delay: None,
117            id_reuse_policy: IdReusePolicy::Default,
118            retry_policy: None,
119            config_override: None,
120            ignore_already_started: false,
121            request_id: None,
122            attributes: Vec::new(),
123        }
124    }
125
126    /// Sets Dex's durable soft timeout. A zero duration disables it.
127    pub fn timeout(mut self, value: Duration) -> Self {
128        self.timeout = Some(value);
129        self
130    }
131
132    /// Sets the action taken when the positive Flow timeout expires.
133    ///
134    /// Handler is rejected before start when the registered Flow has no timeout handler.
135    pub fn timeout_policy(mut self, value: FlowTimeoutPolicy) -> Self {
136        self.timeout_policy = value;
137        self
138    }
139
140    /// Delays the first Step after the server accepts the Flow.
141    pub fn start_delay(mut self, value: Duration) -> Self {
142        self.start_delay = Some(value);
143        self
144    }
145
146    /// Sets how the server handles an existing Flow with the same ID.
147    pub fn id_reuse_policy(mut self, value: IdReusePolicy) -> Self {
148        self.id_reuse_policy = value;
149        self
150    }
151
152    /// Sets the whole-Flow retry policy after a terminal failure.
153    pub fn retry_policy(mut self, value: RetryPolicy) -> Self {
154        self.retry_policy = Some(value);
155        self
156    }
157
158    /// Adds an initial value for a declared Attribute.
159    pub fn initial_attribute<T: Value>(mut self, attribute: &Attribute<T>, value: T) -> Self {
160        self.attributes
161            .push(InitialAttribute::for_attribute(attribute, value));
162        self
163    }
164
165    /// Adds an initial value for one Attribute-map instance.
166    /// Slash is prohibited in instance keys because it is a reserved character.
167    pub fn initial_attribute_map<T: Value>(
168        mut self,
169        attribute: &AttributeMap<T>,
170        instance: &str,
171        value: T,
172    ) -> Self {
173        self.attributes.push(InitialAttribute::for_attribute_map(
174            attribute, instance, value,
175        ));
176        self
177    }
178
179    /// Overrides the registered Flow configuration for this execution.
180    pub fn config_override(mut self, value: FlowConfig) -> Self {
181        self.config_override = Some(value);
182        self
183    }
184
185    /// Returns the existing run ID instead of an error when the Flow already exists.
186    pub fn ignore_already_started(mut self, value: bool) -> Self {
187        self.ignore_already_started = value;
188        self
189    }
190
191    /// Sets the idempotency key for this start request.
192    pub fn request_id(mut self, value: impl Into<String>) -> Self {
193        self.request_id = Some(value.into());
194        self
195    }
196}
197
198impl Default for StartFlowOptions {
199    fn default() -> Self {
200        Self::new()
201    }
202}
203
204#[cfg(test)]
205mod tests {
206    use super::StartFlowOptions;
207    use crate::{Attribute, AttributeMap};
208
209    #[test]
210    fn initial_attributes_retain_sync_configuration() {
211        let plain = Attribute::<String>::new("plain");
212        let synced = Attribute::<String>::new("synced").sync_to_attribute_store();
213        let synced_map = AttributeMap::<String>::new("map").sync_to_attribute_store();
214        let options = StartFlowOptions::new()
215            .initial_attribute(&plain, "plain".to_string())
216            .initial_attribute(&synced, "synced".to_string())
217            .initial_attribute_map(&synced_map, "tenant-1", "mapped".to_string());
218
219        assert!(options.attributes[0].sync_config.is_none());
220        assert_eq!(
221            options.attributes[1]
222                .sync_config
223                .as_ref()
224                .map(|config| config.enabled),
225            Some(true)
226        );
227        assert_eq!(
228            options.attributes[2]
229                .sync_config
230                .as_ref()
231                .map(|config| config.enabled),
232            Some(true)
233        );
234    }
235}
dex_sdk/start_flow_options.rs

dex_sdk/
start_flow_options.rs
