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) cron_schedule: Option<String>,
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            start_delay: None,
118            id_reuse_policy: IdReusePolicy::Default,
119            cron_schedule: None,
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    /// Delays the first Step after the server accepts the Flow.
143    pub fn start_delay(mut self, value: Duration) -> Self {
144        self.start_delay = Some(value);
145        self
146    }
147
148    /// Sets how the server handles an existing Flow with the same ID.
149    pub fn id_reuse_policy(mut self, value: IdReusePolicy) -> Self {
150        self.id_reuse_policy = value;
151        self
152    }
153
154    /// Sets the server cron expression used to schedule recurring runs.
155    pub fn cron_schedule(mut self, value: impl Into<String>) -> Self {
156        self.cron_schedule = Some(value.into());
157        self
158    }
159
160    /// Sets the whole-Flow retry policy after a terminal failure.
161    pub fn retry_policy(mut self, value: RetryPolicy) -> Self {
162        self.retry_policy = Some(value);
163        self
164    }
165
166    /// Adds an initial value for a declared Attribute.
167    pub fn initial_attribute<T: Value>(mut self, attribute: &Attribute<T>, value: T) -> Self {
168        self.attributes
169            .push(InitialAttribute::for_attribute(attribute, value));
170        self
171    }
172
173    /// Adds an initial value for one Attribute-map instance.
174    pub fn initial_attribute_map<T: Value>(
175        mut self,
176        attribute: &AttributeMap<T>,
177        instance: &str,
178        value: T,
179    ) -> Self {
180        self.attributes.push(InitialAttribute::for_attribute_map(
181            attribute, instance, value,
182        ));
183        self
184    }
185
186    /// Overrides the registered Flow configuration for this execution.
187    pub fn config_override(mut self, value: FlowConfig) -> Self {
188        self.config_override = Some(value);
189        self
190    }
191
192    /// Returns the existing run ID instead of an error when the Flow already exists.
193    pub fn ignore_already_started(mut self, value: bool) -> Self {
194        self.ignore_already_started = value;
195        self
196    }
197
198    /// Sets the idempotency key for this start request.
199    pub fn request_id(mut self, value: impl Into<String>) -> Self {
200        self.request_id = Some(value.into());
201        self
202    }
203}
204
205impl Default for StartFlowOptions {
206    fn default() -> Self {
207        Self::new()
208    }
209}
210
211#[cfg(test)]
212mod tests {
213    use super::StartFlowOptions;
214    use crate::{Attribute, AttributeMap};
215
216    #[test]
217    fn initial_attributes_retain_sync_configuration() {
218        let plain = Attribute::<String>::new("plain");
219        let synced = Attribute::<String>::new("synced").sync_to_attribute_store();
220        let synced_map = AttributeMap::<String>::new("map").sync_to_attribute_store();
221        let options = StartFlowOptions::new()
222            .initial_attribute(&plain, "plain".to_string())
223            .initial_attribute(&synced, "synced".to_string())
224            .initial_attribute_map(&synced_map, "tenant-1", "mapped".to_string());
225
226        assert!(options.attributes[0].sync_config.is_none());
227        assert_eq!(
228            options.attributes[1]
229                .sync_config
230                .as_ref()
231                .map(|config| config.enabled),
232            Some(true)
233        );
234        assert_eq!(
235            options.attributes[2]
236                .sync_config
237                .as_ref()
238                .map(|config| config.enabled),
239            Some(true)
240        );
241    }
242}
dex_sdk/start_flow_options.rs

dex_sdk/
start_flow_options.rs
