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

dex_sdk/
start_flow_options.rs
