dex_sdk/
start_flow_options.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::time::Duration;
10
11use std::sync::Arc;
12
13use dex_protocol::dex::AttributeWrite;
14
15use crate::attribute::{
16    AttributeWriteSettings, require_registered_hard_delete_setting,
17    validate_hard_delete_attribute_write,
18};
19use crate::client::invalid;
20use crate::persistence::PersistenceDefinition;
21use crate::registry::physical_name;
22use crate::step::{ErasedValue, TypedValue};
23use crate::{
24    Attribute, AttributeMap, FlowConfig, FlowTimeoutHandlerOptions, RetryPolicy, SdkResult, Value,
25};
26
27#[derive(Clone, Copy, Debug, Eq, PartialEq)]
28/// Controls whether [`crate::Client::start_flow_with_options`] may reuse a Flow ID.
29pub enum IdReusePolicy {
30    /// Uses the Dex server's configured reuse policy.
31    Default,
32    /// Allows reuse only when the previous run failed.
33    AllowIfPreviousFailed,
34    /// Allows reuse when no run with the ID is currently active.
35    AllowIfNotRunning,
36    /// Rejects reuse regardless of the previous run status.
37    Disallow,
38    /// Terminates an active run before starting the replacement.
39    TerminateIfRunning,
40}
41
42#[derive(Clone, Copy, Debug, Eq, PartialEq)]
43/// Controls how Dex responds when a positive soft Flow timeout expires.
44pub enum FlowTimeoutPolicy {
45    /// Uses the Flow's registered timeout handler when present, and fails otherwise.
46    Default,
47    /// Fails with [`crate::FlowErrorType::FlowTimeout`] and permits Flow retries.
48    Fail,
49    /// Cancels without retrying the Flow.
50    Cancel,
51    /// Invokes the registered [`crate::FlowTimeoutHandler`] as one retryable logical execution.
52    Handler,
53}
54
55#[derive(Clone)]
56/// Configures a new Flow execution.
57///
58/// All optional fields are absent by default. Initial Attribute values are encoded when the start
59/// request is sent. Use a stable request ID to make retries of the same logical request idempotent.
60///
61/// A stable request ID alone does not make a retry succeed. Set
62/// [`StartFlowOptions::request_id`] and [`StartFlowOptions::ignore_already_started`] together, and
63/// reuse the same values on every retry of one logical start, for example after a lost response.
64///
65/// # Examples
66///
67/// ```
68/// use dex_sdk::{Attribute, IdReusePolicy, StartFlowOptions};
69/// use std::time::Duration;
70///
71/// let customer = Attribute::<String>::new("customer");
72/// let options = StartFlowOptions::new()
73///     .timeout(Duration::from_secs(3_600))
74///     .id_reuse_policy(IdReusePolicy::AllowIfNotRunning)
75///     .initial_attribute(&customer, "customer-42".to_owned())
76///     .request_id("create-order-42");
77/// ```
78///
79/// Retry-safe start options:
80///
81/// ```
82/// use dex_sdk::StartFlowOptions;
83///
84/// let order_id = "42";
85/// let options = StartFlowOptions::new()
86///     .ignore_already_started(true)
87///     .request_id(format!("start-order-{order_id}"));
88/// ```
89pub struct StartFlowOptions {
90    pub(crate) timeout: Option<Duration>,
91    pub(crate) timeout_policy: FlowTimeoutPolicy,
92    pub(crate) timeout_handler_options: Option<FlowTimeoutHandlerOptions>,
93    pub(crate) start_delay: Option<Duration>,
94    pub(crate) id_reuse_policy: IdReusePolicy,
95    pub(crate) retry_policy: Option<RetryPolicy>,
96    pub(crate) config_override: Option<FlowConfig>,
97    pub(crate) ignore_already_started: bool,
98    pub(crate) request_id: Option<String>,
99    pub(crate) attributes: Vec<InitialAttribute>,
100}
101
102#[derive(Clone)]
103pub(crate) struct InitialAttribute {
104    pub(crate) key: String,
105    pub(crate) attribute_name: String,
106    pub(crate) value: Arc<dyn ErasedValue>,
107    pub(crate) write_settings: AttributeWriteSettings,
108}
109
110impl InitialAttribute {
111    pub(crate) fn for_attribute<T: Value>(attribute: &Attribute<T>, value: T) -> Self {
112        Self {
113            key: attribute.name().to_string(),
114            attribute_name: attribute.name().to_string(),
115            value: Arc::new(TypedValue(value)),
116            write_settings: attribute.write_settings(),
117        }
118    }
119
120    pub(crate) fn for_attribute_map<T: Value>(
121        attribute: &AttributeMap<T>,
122        instance: &str,
123        value: T,
124    ) -> Self {
125        Self {
126            key: physical_name(attribute.name(), instance),
127            attribute_name: attribute.name().to_string(),
128            value: Arc::new(TypedValue(value)),
129            write_settings: attribute.write_settings(),
130        }
131    }
132
133    /// Returns the encoded write after checking it against the registered `definition`.
134    ///
135    /// Returns [`crate::SdkError::InvalidArgument`] for a hard-delete mismatch or an invalid
136    /// hard-delete combination or value. Returns [`crate::SdkError::ValueMapping`] when the value
137    /// cannot be encoded.
138    pub(crate) fn encode_attribute_write(
139        &self,
140        definition: &PersistenceDefinition,
141    ) -> SdkResult<AttributeWrite> {
142        require_registered_hard_delete_setting(
143            &self.attribute_name,
144            definition.is_hard_delete_on_null,
145            self.write_settings.is_hard_delete_on_null,
146        )
147        .map_err(invalid)?;
148        let write = self
149            .write_settings
150            .clone()
151            .attribute_write(self.key.clone(), self.value.encode()?);
152        validate_hard_delete_attribute_write(&write).map_err(invalid)?;
153        Ok(write)
154    }
155}
156
157impl StartFlowOptions {
158    /// Creates start options that preserve all server defaults.
159    pub fn new() -> Self {
160        Self {
161            timeout: None,
162            timeout_policy: FlowTimeoutPolicy::Default,
163            timeout_handler_options: None,
164            start_delay: None,
165            id_reuse_policy: IdReusePolicy::Default,
166            retry_policy: None,
167            config_override: None,
168            ignore_already_started: false,
169            request_id: None,
170            attributes: Vec::new(),
171        }
172    }
173
174    /// Sets Dex's durable soft timeout. A zero duration disables it.
175    pub fn timeout(mut self, value: Duration) -> Self {
176        self.timeout = Some(value);
177        self
178    }
179
180    /// Sets the action taken when the positive Flow timeout expires.
181    ///
182    /// Handler is rejected before start when the registered Flow has no timeout handler.
183    pub fn timeout_policy(mut self, value: FlowTimeoutPolicy) -> Self {
184        self.timeout_policy = value;
185        self
186    }
187
188    /// Configures execution and selective state loading for the Flow timeout handler.
189    ///
190    /// Dex rejects this option unless the Flow has a positive timeout and resolves to
191    /// [`FlowTimeoutPolicy::Handler`].
192    pub fn timeout_handler_options(mut self, value: FlowTimeoutHandlerOptions) -> Self {
193        self.timeout_handler_options = Some(value);
194        self
195    }
196
197    /// Delays the first Step after the server accepts the Flow.
198    pub fn start_delay(mut self, value: Duration) -> Self {
199        self.start_delay = Some(value);
200        self
201    }
202
203    /// Sets how the server handles an existing Flow with the same ID.
204    pub fn id_reuse_policy(mut self, value: IdReusePolicy) -> Self {
205        self.id_reuse_policy = value;
206        self
207    }
208
209    /// Sets the whole-Flow retry policy after a terminal failure.
210    pub fn retry_policy(mut self, value: RetryPolicy) -> Self {
211        self.retry_policy = Some(value);
212        self
213    }
214
215    /// Adds an initial value for a declared Attribute.
216    pub fn initial_attribute<T: Value>(mut self, attribute: &Attribute<T>, value: T) -> Self {
217        self.attributes
218            .push(InitialAttribute::for_attribute(attribute, value));
219        self
220    }
221
222    /// Adds an initial value for one Attribute-map instance.
223    /// Slash is prohibited in instance keys because it is a reserved character.
224    pub fn initial_attribute_map<T: Value>(
225        mut self,
226        attribute: &AttributeMap<T>,
227        instance: &str,
228        value: T,
229    ) -> Self {
230        self.attributes.push(InitialAttribute::for_attribute_map(
231            attribute, instance, value,
232        ));
233        self
234    }
235
236    /// Overrides the registered Flow configuration for this execution.
237    pub fn config_override(mut self, value: FlowConfig) -> Self {
238        self.config_override = Some(value);
239        self
240    }
241
242    /// Returns the existing run ID instead of an error when the Flow already exists.
243    ///
244    /// It applies only when [`StartFlowOptions::request_id`] matches the request ID that started
245    /// the existing run. A different or generated request ID still returns
246    /// [`SdkError::FlowAlreadyStarted`](crate::SdkError::FlowAlreadyStarted). See
247    /// [`StartFlowOptions`] for an example that sets both options.
248    pub fn ignore_already_started(mut self, value: bool) -> Self {
249        self.ignore_already_started = value;
250        self
251    }
252
253    /// Sets the idempotency key for this start request.
254    ///
255    /// When omitted, the SDK generates a new request ID for every call. A retried start returns the
256    /// existing run only when [`StartFlowOptions::ignore_already_started`] is `true` and this
257    /// request ID matches the request that started that run. Without that option, the retry returns
258    /// [`SdkError::FlowAlreadyStarted`](crate::SdkError::FlowAlreadyStarted) even when the first
259    /// attempt created the Flow. The match applies only when the ID reuse policy rejects the start.
260    /// A policy that permits another run starts a new run instead. Dex compares only request IDs,
261    /// not input or other options, so never reuse a request ID for a different logical start. See
262    /// [`StartFlowOptions`] for an example that sets both options.
263    pub fn request_id(mut self, value: impl Into<String>) -> Self {
264        self.request_id = Some(value.into());
265        self
266    }
267}
268
269impl Default for StartFlowOptions {
270    fn default() -> Self {
271        Self::new()
272    }
273}
274
275#[cfg(test)]
276mod tests {
277    use super::StartFlowOptions;
278    use crate::persistence::{PersistenceDefinition, PersistenceKind};
279    use crate::{Attribute, AttributeMap, SdkError};
280
281    #[test]
282    fn initial_attributes_retain_sync_configuration() {
283        let plain = Attribute::<String>::new("plain");
284        let synced = Attribute::<String>::new("synced").sync_to_attribute_store();
285        let synced_map = AttributeMap::<String>::new("map").sync_to_attribute_store();
286        let options = StartFlowOptions::new()
287            .initial_attribute(&plain, "plain".to_string())
288            .initial_attribute(&synced, "synced".to_string())
289            .initial_attribute_map(&synced_map, "tenant-1", "mapped".to_string());
290
291        assert!(options.attributes[0].write_settings.sync_config.is_none());
292        assert_eq!(
293            options.attributes[1]
294                .write_settings
295                .sync_config
296                .as_ref()
297                .map(|config| config.enabled),
298            Some(true)
299        );
300        assert_eq!(
301            options.attributes[2]
302                .write_settings
303                .sync_config
304                .as_ref()
305                .map(|config| config.enabled),
306            Some(true)
307        );
308    }
309
310    #[derive(serde::Deserialize, serde::Serialize)]
311    struct NumericNewtype(u64);
312
313    #[test]
314    fn initial_hard_delete_attributes_set_the_flag_and_validate_values() {
315        let secret = Attribute::<String>::new("secret").hard_delete_on_null();
316        let secrets = AttributeMap::<Option<String>>::new("secrets").hard_delete_on_null();
317        let numeric = Attribute::<NumericNewtype>::new("numeric").hard_delete_on_null();
318        let plain_secret = Attribute::<String>::new("secret");
319        let options = StartFlowOptions::new()
320            .initial_attribute(&secret, "key-1".to_string())
321            .initial_attribute_map(&secrets, "tenant-1", None)
322            .initial_attribute(&numeric, NumericNewtype(7))
323            .initial_attribute(&plain_secret, "leak".to_string());
324        let secret_definition = PersistenceDefinition {
325            is_hard_delete_on_null: true,
326            ..PersistenceDefinition::new("secret", PersistenceKind::Attribute)
327        };
328        let secrets_definition = PersistenceDefinition {
329            is_hard_delete_on_null: true,
330            ..PersistenceDefinition::new("secrets", PersistenceKind::AttributeMap)
331        };
332        let numeric_definition = PersistenceDefinition {
333            is_hard_delete_on_null: true,
334            ..PersistenceDefinition::new("numeric", PersistenceKind::Attribute)
335        };
336
337        let write = options.attributes[0]
338            .encode_attribute_write(&secret_definition)
339            .expect("string write");
340        assert!(write.is_hard_delete_on_null);
341        assert!(write.index_config.is_none() && write.sync_config.is_none());
342        let deletion = options.attributes[1]
343            .encode_attribute_write(&secrets_definition)
344            .expect("null write");
345        assert!(deletion.is_hard_delete_on_null);
346        assert_eq!(deletion.key, "secrets/tenant-1");
347        for (result, expected) in [
348            (
349                options.attributes[2].encode_attribute_write(&numeric_definition),
350                "numeric and Boolean values are rejected",
351            ),
352            (
353                options.attributes[3].encode_attribute_write(&secret_definition),
354                "does not match its registered definition",
355            ),
356            (
357                options.attributes[0].encode_attribute_write(&PersistenceDefinition::new(
358                    "secret",
359                    PersistenceKind::Attribute,
360                )),
361                "does not match its registered definition",
362            ),
363        ] {
364            let Err(SdkError::InvalidArgument { message }) = result else {
365                panic!("expected InvalidArgument");
366            };
367            assert!(message.contains(expected), "{message}");
368        }
369    }
370}
dex_sdk/start_flow_options.rs

dex_sdk/
start_flow_options.rs
