dex_sdk/
step_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::marker::PhantomData;
10use std::time::Duration;
11
12use crate::attribute::AttributeLock;
13use crate::attribute::AttributeMapLoad;
14use crate::channel::{ChannelLoad, ChannelMapLoad};
15use crate::{AttributeMap, Channel, ChannelMap, RetryPolicy, Step, Value};
16
17/// Configures one Step's handler execution and persistence behavior.
18///
19/// New options preserve server timeout, retry, and durability defaults. `wait_for` failures fail the
20/// Flow by default. Attribute locks apply only to the matching handler call. The type system checks
21/// that an execute-failure recovery Step accepts the same input type.
22///
23/// # Examples
24///
25/// ```
26/// use dex_sdk::{Attribute, RetryPolicy, StepOptions, WaitForFailurePolicy};
27/// use std::time::Duration;
28///
29/// let balance = Attribute::<i64>::new("balance");
30/// let options = StepOptions::<String>::new()
31///     .execute_method_timeout(Duration::from_secs(30))
32///     .execute_retry(RetryPolicy::new().maximum_attempts(3))
33///     .wait_for_failure(WaitForFailurePolicy::Proceed)
34///     .execute_lock(balance.lock());
35/// ```
36pub struct StepOptions<Input> {
37    pub(crate) wait_for_method_timeout: Option<Duration>,
38    pub(crate) execute_method_timeout: Option<Duration>,
39    pub(crate) heartbeat_timeout: Option<Duration>,
40    pub(crate) wait_for_retry: Option<RetryPolicy>,
41    pub(crate) execute_retry: Option<RetryPolicy>,
42    pub(crate) wait_for_failure: WaitForFailurePolicy,
43    pub(crate) wait_for_durability: StepDurability,
44    pub(crate) execute_durability: StepDurability,
45    pub(crate) wait_for_locks: Vec<AttributeLock>,
46    pub(crate) execute_locks: Vec<AttributeLock>,
47    pub(crate) wait_for_load_attribute_maps: Vec<AttributeMapLoad>,
48    pub(crate) wait_for_load_channels: Vec<ChannelLoad>,
49    pub(crate) wait_for_load_channel_maps: Vec<ChannelMapLoad>,
50    pub(crate) execute_load_attribute_maps: Vec<AttributeMapLoad>,
51    pub(crate) execute_load_channels: Vec<ChannelLoad>,
52    pub(crate) execute_load_channel_maps: Vec<ChannelMapLoad>,
53    pub(crate) execute_failure_step: Option<&'static str>,
54    marker: PhantomData<fn(Input)>,
55}
56
57#[derive(Clone)]
58pub(crate) struct ErasedStepOptions {
59    pub(crate) wait_for_method_timeout: Option<Duration>,
60    pub(crate) execute_method_timeout: Option<Duration>,
61    pub(crate) heartbeat_timeout: Option<Duration>,
62    pub(crate) wait_for_retry: Option<RetryPolicy>,
63    pub(crate) execute_retry: Option<RetryPolicy>,
64    pub(crate) wait_for_failure: WaitForFailurePolicy,
65    pub(crate) wait_for_durability: StepDurability,
66    pub(crate) execute_durability: StepDurability,
67    pub(crate) wait_for_locks: Vec<AttributeLock>,
68    pub(crate) execute_locks: Vec<AttributeLock>,
69    pub(crate) wait_for_load_attribute_maps: Vec<AttributeMapLoad>,
70    pub(crate) wait_for_load_channels: Vec<ChannelLoad>,
71    pub(crate) wait_for_load_channel_maps: Vec<ChannelMapLoad>,
72    pub(crate) execute_load_attribute_maps: Vec<AttributeMapLoad>,
73    pub(crate) execute_load_channels: Vec<ChannelLoad>,
74    pub(crate) execute_load_channel_maps: Vec<ChannelMapLoad>,
75    pub(crate) execute_failure_step: Option<&'static str>,
76}
77
78impl<Input> From<StepOptions<Input>> for ErasedStepOptions {
79    fn from(options: StepOptions<Input>) -> Self {
80        Self {
81            wait_for_method_timeout: options.wait_for_method_timeout,
82            execute_method_timeout: options.execute_method_timeout,
83            heartbeat_timeout: options.heartbeat_timeout,
84            wait_for_retry: options.wait_for_retry,
85            execute_retry: options.execute_retry,
86            wait_for_failure: options.wait_for_failure,
87            wait_for_durability: options.wait_for_durability,
88            execute_durability: options.execute_durability,
89            wait_for_locks: options.wait_for_locks,
90            execute_locks: options.execute_locks,
91            wait_for_load_attribute_maps: options.wait_for_load_attribute_maps,
92            wait_for_load_channels: options.wait_for_load_channels,
93            wait_for_load_channel_maps: options.wait_for_load_channel_maps,
94            execute_load_attribute_maps: options.execute_load_attribute_maps,
95            execute_load_channels: options.execute_load_channels,
96            execute_load_channel_maps: options.execute_load_channel_maps,
97            execute_failure_step: options.execute_failure_step,
98        }
99    }
100}
101
102impl<Input: Value> StepOptions<Input> {
103    /// Creates options that preserve server defaults and fail on exhausted `wait_for` retries.
104    pub fn new() -> Self {
105        Self {
106            wait_for_method_timeout: None,
107            execute_method_timeout: None,
108            heartbeat_timeout: None,
109            wait_for_retry: None,
110            execute_retry: None,
111            wait_for_failure: WaitForFailurePolicy::FailFlow,
112            wait_for_durability: StepDurability::Default,
113            execute_durability: StepDurability::Default,
114            wait_for_locks: Vec::new(),
115            execute_locks: Vec::new(),
116            wait_for_load_attribute_maps: Vec::new(),
117            wait_for_load_channels: Vec::new(),
118            wait_for_load_channel_maps: Vec::new(),
119            execute_load_attribute_maps: Vec::new(),
120            execute_load_channels: Vec::new(),
121            execute_load_channel_maps: Vec::new(),
122            execute_failure_step: None,
123            marker: PhantomData,
124        }
125    }
126
127    /// Sets the maximum duration of one `wait_for` handler attempt.
128    pub fn wait_for_method_timeout(mut self, value: Duration) -> Self {
129        self.wait_for_method_timeout = Some(value);
130        self
131    }
132
133    /// Sets the maximum duration of one `execute` handler attempt.
134    pub fn execute_method_timeout(mut self, value: Duration) -> Self {
135        self.execute_method_timeout = Some(value);
136        self
137    }
138
139    /// Sets the heartbeat timeout for regular `wait_for` and `execute` activities.
140    ///
141    /// Zero selects the server default of one minute. Positive values must be whole seconds and at
142    /// least the server-configured minimum, which defaults to ten seconds. Local activities ignore
143    /// this option; an asynchronous fallback to a regular activity uses it.
144    pub fn heartbeat_timeout(mut self, value: Duration) -> Self {
145        self.heartbeat_timeout = Some(value);
146        self
147    }
148
149    /// Sets the retry policy for failed `wait_for` attempts.
150    pub fn wait_for_retry(mut self, value: RetryPolicy) -> Self {
151        self.wait_for_retry = Some(value);
152        self
153    }
154
155    /// Sets the retry policy for failed `execute` attempts.
156    pub fn execute_retry(mut self, value: RetryPolicy) -> Self {
157        self.execute_retry = Some(value);
158        self
159    }
160
161    /// Selects whether exhausted `wait_for` retries fail or advance the Flow.
162    pub fn wait_for_failure(mut self, value: WaitForFailurePolicy) -> Self {
163        self.wait_for_failure = value;
164        self
165    }
166
167    /// Routes exhausted `execute` failures to a recovery Step with the same input type.
168    pub fn on_execute_failure_proceed_to<RecoveryStep>(mut self, step: &RecoveryStep) -> Self
169    where
170        RecoveryStep: Step<Input = Input>,
171    {
172        self.execute_failure_step = Some(step.step_type());
173        self
174    }
175
176    /// Overrides persistence durability for writes staged by `wait_for`.
177    pub fn wait_for_durability(mut self, value: StepDurability) -> Self {
178        self.wait_for_durability = value;
179        self
180    }
181
182    /// Overrides persistence durability for writes staged by `execute`.
183    pub fn execute_durability(mut self, value: StepDurability) -> Self {
184        self.execute_durability = value;
185        self
186    }
187
188    /// Adds an Attribute lock held for the `wait_for` invocation.
189    pub fn wait_for_lock(mut self, value: AttributeLock) -> Self {
190        self.wait_for_locks.push(value);
191        self
192    }
193
194    /// Adds an Attribute lock held for the `execute` invocation.
195    pub fn execute_lock(mut self, value: AttributeLock) -> Self {
196        self.execute_locks.push(value);
197        self
198    }
199
200    /// Loads every current instance of `attribute_map` for `wait_for`.
201    pub fn wait_for_load_attribute_map<T>(mut self, attribute_map: &AttributeMap<T>) -> Self {
202        self.wait_for_load_attribute_maps.push(AttributeMapLoad {
203            name: attribute_map.name().to_owned(),
204            instance: None,
205        });
206        self
207    }
208
209    /// Loads one AttributeMap instance for `wait_for`.
210    pub fn wait_for_load_attribute_map_instance(mut self, load: AttributeMapLoad) -> Self {
211        self.wait_for_load_attribute_maps.push(load);
212        self
213    }
214
215    /// Loads one Channel's pending messages for `wait_for`.
216    pub fn wait_for_load_channel<T>(mut self, channel: &Channel<T>) -> Self {
217        self.wait_for_load_channels.push(ChannelLoad {
218            name: channel.name().to_owned(),
219        });
220        self
221    }
222
223    /// Loads every current ChannelMap instance's pending messages for `wait_for`.
224    pub fn wait_for_load_channel_map<T>(mut self, channel_map: &ChannelMap<T>) -> Self {
225        self.wait_for_load_channel_maps.push(ChannelMapLoad {
226            name: channel_map.name().to_owned(),
227            instance: None,
228        });
229        self
230    }
231
232    /// Loads one ChannelMap instance's pending messages for `wait_for`.
233    pub fn wait_for_load_channel_map_instance(mut self, load: ChannelMapLoad) -> Self {
234        self.wait_for_load_channel_maps.push(load);
235        self
236    }
237
238    /// Loads every current instance of `attribute_map` for `execute`.
239    pub fn execute_load_attribute_map<T>(mut self, attribute_map: &AttributeMap<T>) -> Self {
240        self.execute_load_attribute_maps.push(AttributeMapLoad {
241            name: attribute_map.name().to_owned(),
242            instance: None,
243        });
244        self
245    }
246
247    /// Loads one AttributeMap instance for `execute`.
248    pub fn execute_load_attribute_map_instance(mut self, load: AttributeMapLoad) -> Self {
249        self.execute_load_attribute_maps.push(load);
250        self
251    }
252
253    /// Loads one Channel's pending messages for `execute`.
254    pub fn execute_load_channel<T>(mut self, channel: &Channel<T>) -> Self {
255        self.execute_load_channels.push(ChannelLoad {
256            name: channel.name().to_owned(),
257        });
258        self
259    }
260
261    /// Loads every current ChannelMap instance's pending messages for `execute`.
262    pub fn execute_load_channel_map<T>(mut self, channel_map: &ChannelMap<T>) -> Self {
263        self.execute_load_channel_maps.push(ChannelMapLoad {
264            name: channel_map.name().to_owned(),
265            instance: None,
266        });
267        self
268    }
269
270    /// Loads one ChannelMap instance's pending messages for `execute`.
271    pub fn execute_load_channel_map_instance(mut self, load: ChannelMapLoad) -> Self {
272        self.execute_load_channel_maps.push(load);
273        self
274    }
275}
276
277impl<Input: Value> Default for StepOptions<Input> {
278    fn default() -> Self {
279        Self::new()
280    }
281}
282
283#[derive(Clone, Copy, Debug, Eq, PartialEq)]
284/// Selects the terminal behavior after `wait_for` exhausts its retry policy.
285pub enum WaitForFailurePolicy {
286    /// Fails the Flow.
287    FailFlow,
288    /// Skips the failed wait and invokes `execute` with failure visible in [`crate::Context`].
289    Proceed,
290}
291
292#[derive(Clone, Copy, Debug, Eq, PartialEq)]
293/// Controls when staged Step persistence must be durably acknowledged.
294pub enum StepDurability {
295    /// Uses the Flow or server default.
296    Default,
297    /// Waits for durable persistence before completing the handler request.
298    Sync,
299    /// Allows asynchronous persistence after accepting the handler result.
300    Async,
301}
dex_sdk/step_options.rs

dex_sdk/
step_options.rs
