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    /// Sets a delay before the first Step after the server accepts the Flow.
198    ///
199    /// The Server currently supports only a zero delay. It rejects a positive delay with
200    /// INVALID_ARGUMENT. To start later, call [`crate::Client::start_flow`] at the desired time or
201    /// wait for a Timer in the starting Step.
202    pub fn start_delay(mut self, value: Duration) -> Self {
203        self.start_delay = Some(value);
204        self
205    }
206
207    /// Sets how the server handles an existing Flow with the same ID.
208    pub fn id_reuse_policy(mut self, value: IdReusePolicy) -> Self {
209        self.id_reuse_policy = value;
210        self
211    }
212
213    /// Sets a whole-Flow retry policy for the request.
214    ///
215    /// The Server does not support whole-Flow retry. It rejects a request that sets this option
216    /// with INVALID_ARGUMENT. Configure [`crate::RetryPolicy`] on Step methods through
217    /// [`crate::StepOptions`] instead.
218    pub fn retry_policy(mut self, value: RetryPolicy) -> Self {
219        self.retry_policy = Some(value);
220        self
221    }
222
223    /// Adds an initial value for a declared Attribute.
224    pub fn initial_attribute<T: Value>(mut self, attribute: &Attribute<T>, value: T) -> Self {
225        self.attributes
226            .push(InitialAttribute::for_attribute(attribute, value));
227        self
228    }
229
230    /// Adds an initial value for one Attribute-map instance.
231    /// Slash is prohibited in instance keys because it is a reserved character.
232    pub fn initial_attribute_map<T: Value>(
233        mut self,
234        attribute: &AttributeMap<T>,
235        instance: &str,
236        value: T,
237    ) -> Self {
238        self.attributes.push(InitialAttribute::for_attribute_map(
239            attribute, instance, value,
240        ));
241        self
242    }
243
244    /// Overrides the registered Flow configuration for this execution.
245    pub fn config_override(mut self, value: FlowConfig) -> Self {
246        self.config_override = Some(value);
247        self
248    }
249
250    /// Returns the existing run ID instead of an error when the Flow already exists.
251    ///
252    /// It applies only when [`StartFlowOptions::request_id`] matches the request ID that started
253    /// the existing run. A different or generated request ID still returns
254    /// [`SdkError::FlowAlreadyStarted`](crate::SdkError::FlowAlreadyStarted). See
255    /// [`StartFlowOptions`] for an example that sets both options.
256    pub fn ignore_already_started(mut self, value: bool) -> Self {
257        self.ignore_already_started = value;
258        self
259    }
260
261    /// Sets the idempotency key for this start request.
262    ///
263    /// When omitted, the SDK generates a new request ID for every call. A retried start returns the
264    /// existing run only when [`StartFlowOptions::ignore_already_started`] is `true` and this
265    /// request ID matches the request that started that run. Without that option, the retry returns
266    /// [`SdkError::FlowAlreadyStarted`](crate::SdkError::FlowAlreadyStarted) even when the first
267    /// attempt created the Flow. The match applies only when the ID reuse policy rejects the start.
268    /// A policy that permits another run starts a new run instead. Dex compares only request IDs,
269    /// not input or other options, so never reuse a request ID for a different logical start. See
270    /// [`StartFlowOptions`] for an example that sets both options.
271    pub fn request_id(mut self, value: impl Into<String>) -> Self {
272        self.request_id = Some(value.into());
273        self
274    }
275}
276
277impl Default for StartFlowOptions {
278    fn default() -> Self {
279        Self::new()
280    }
281}
282
283#[cfg(test)]
284mod tests {
285    use super::StartFlowOptions;
286    use crate::persistence::{PersistenceDefinition, PersistenceKind};
287    use crate::{Attribute, AttributeMap, SdkError};
288
289    #[test]
290    fn initial_attributes_retain_sync_configuration() {
291        let plain = Attribute::<String>::new("plain");
292        let synced = Attribute::<String>::new("synced").sync_to_attribute_store();
293        let synced_map = AttributeMap::<String>::new("map").sync_to_attribute_store();
294        let options = StartFlowOptions::new()
295            .initial_attribute(&plain, "plain".to_string())
296            .initial_attribute(&synced, "synced".to_string())
297            .initial_attribute_map(&synced_map, "tenant-1", "mapped".to_string());
298
299        assert!(options.attributes[0].write_settings.sync_config.is_none());
300        assert_eq!(
301            options.attributes[1]
302                .write_settings
303                .sync_config
304                .as_ref()
305                .map(|config| config.enabled),
306            Some(true)
307        );
308        assert_eq!(
309            options.attributes[2]
310                .write_settings
311                .sync_config
312                .as_ref()
313                .map(|config| config.enabled),
314            Some(true)
315        );
316    }
317
318    #[derive(serde::Deserialize, serde::Serialize)]
319    struct NumericNewtype(u64);
320
321    #[test]
322    fn initial_hard_delete_attributes_set_the_flag_and_validate_values() {
323        let secret = Attribute::<String>::new("secret").hard_delete_on_null();
324        let secrets = AttributeMap::<Option<String>>::new("secrets").hard_delete_on_null();
325        let numeric = Attribute::<NumericNewtype>::new("numeric").hard_delete_on_null();
326        let plain_secret = Attribute::<String>::new("secret");
327        let options = StartFlowOptions::new()
328            .initial_attribute(&secret, "key-1".to_string())
329            .initial_attribute_map(&secrets, "tenant-1", None)
330            .initial_attribute(&numeric, NumericNewtype(7))
331            .initial_attribute(&plain_secret, "leak".to_string());
332        let secret_definition = PersistenceDefinition {
333            is_hard_delete_on_null: true,
334            ..PersistenceDefinition::new("secret", PersistenceKind::Attribute)
335        };
336        let secrets_definition = PersistenceDefinition {
337            is_hard_delete_on_null: true,
338            ..PersistenceDefinition::new("secrets", PersistenceKind::AttributeMap)
339        };
340        let numeric_definition = PersistenceDefinition {
341            is_hard_delete_on_null: true,
342            ..PersistenceDefinition::new("numeric", PersistenceKind::Attribute)
343        };
344
345        let write = options.attributes[0]
346            .encode_attribute_write(&secret_definition)
347            .expect("string write");
348        assert!(write.is_hard_delete_on_null);
349        assert!(write.index_config.is_none() && write.sync_config.is_none());
350        let deletion = options.attributes[1]
351            .encode_attribute_write(&secrets_definition)
352            .expect("null write");
353        assert!(deletion.is_hard_delete_on_null);
354        assert_eq!(deletion.key, "secrets/tenant-1");
355        for (result, expected) in [
356            (
357                options.attributes[2].encode_attribute_write(&numeric_definition),
358                "numeric and Boolean values are rejected",
359            ),
360            (
361                options.attributes[3].encode_attribute_write(&secret_definition),
362                "does not match its registered definition",
363            ),
364            (
365                options.attributes[0].encode_attribute_write(&PersistenceDefinition::new(
366                    "secret",
367                    PersistenceKind::Attribute,
368                )),
369                "does not match its registered definition",
370            ),
371        ] {
372            let Err(SdkError::InvalidArgument { message }) = result else {
373                panic!("expected InvalidArgument");
374            };
375            assert!(message.contains(expected), "{message}");
376        }
377    }
378}
dex_sdk/start_flow_options.rs

dex_sdk/
start_flow_options.rs
