dex_sdk/
step.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::marker::PhantomData;
10
11use dex_protocol::dex::Value as ProtoValue;
12
13use crate::channel::ChannelGuard;
14use crate::step_options::ErasedStepOptions;
15use crate::value_mapper;
16use crate::{Context, HandlerResult, SdkResult, StepOptions, Value, Wait, short_type_name};
17
18/// Defines one durable state transition in a Flow.
19///
20/// Dex calls [`Self::wait_for`] to obtain durable conditions, then calls [`Self::execute`] after the
21/// wait is satisfied. The default `wait_for` skips immediately, making execute-only Steps concise.
22/// Handler code may read and stage persistence changes through [`Context`].
23///
24/// # Examples
25///
26/// ```
27/// use dex_sdk::{Context, HandlerResult, Step, StepDecision};
28///
29/// struct CompleteOrder;
30///
31/// impl Step for CompleteOrder {
32///     type Input = String;
33///
34///     fn execute(
35///         &self,
36///         _context: &mut Context,
37///         order_id: Self::Input,
38///     ) -> HandlerResult<StepDecision> {
39///         Ok(StepDecision::graceful_complete(order_id))
40///     }
41/// }
42/// ```
43pub trait Step: Send + Sync + 'static {
44    /// Value decoded for both `wait_for` and `execute`.
45    type Input: Value;
46
47    /// Produces the next movement or terminal Flow decision.
48    ///
49    /// `context` represents one isolated invocation and `input` is a newly decoded owned value.
50    /// Returning [`HandlerError`](crate::HandlerError) activates the configured execute retry and
51    /// failure behavior.
52    fn execute(&self, context: &mut Context, input: Self::Input) -> HandlerResult<StepDecision>;
53
54    /// Returns the durable conditions Dex must satisfy before `execute`.
55    ///
56    /// The default returns [`Wait::skip_immediately`]. Returning an error activates the configured
57    /// `wait_for` retry and failure behavior.
58    fn wait_for(&self, _context: &mut Context, _input: Self::Input) -> HandlerResult<Wait> {
59        Ok(Wait::skip_immediately())
60    }
61
62    /// Returns the stable Step type sent to Dex.
63    ///
64    /// The default is the final Rust type-name segment.
65    fn step_type(&self) -> &'static str {
66        short_type_name::<Self>()
67    }
68
69    /// Returns timeouts, retries, locks, durability, and failure behavior for this Step.
70    ///
71    /// The default preserves server defaults and fails the Flow when `wait_for` exhausts retries.
72    fn options(&self) -> StepOptions<Self::Input> {
73        StepOptions::new()
74    }
75}
76
77/// Collects a Flow's starting Step and remaining Step definitions.
78///
79/// The lifetime ties registered definitions to Step values owned by the Flow. A list may contain at
80/// most one starting Step; [`Self::and`] adds non-starting Steps with arbitrary input types.
81pub struct StepList<'a, StartInput> {
82    definitions: Vec<StepDefinition<'a>>,
83    marker: PhantomData<fn(StartInput)>,
84}
85
86impl<'a, StartInput: Value> StepList<'a, StartInput> {
87    /// Creates a list without a starting Step.
88    pub fn empty() -> Self {
89        Self {
90            definitions: Vec::new(),
91            marker: PhantomData,
92        }
93    }
94
95    /// Creates a list whose starting Step accepts the Flow's start input.
96    pub fn start<StartStep>(step: &'a StartStep) -> Self
97    where
98        StartStep: Step<Input = StartInput>,
99    {
100        Self {
101            definitions: vec![StepDefinition::new(step, true)],
102            marker: PhantomData,
103        }
104    }
105
106    /// Adds a non-starting Step and returns the updated list.
107    pub fn and<OtherStep>(mut self, step: &'a OtherStep) -> Self
108    where
109        OtherStep: Step,
110    {
111        self.definitions.push(StepDefinition::new(step, false));
112        self
113    }
114
115    pub(crate) fn into_definitions(self) -> Vec<RegisteredStep> {
116        self.definitions
117            .into_iter()
118            .map(|definition| RegisteredStep {
119                name: definition.name,
120                starting: definition.starting,
121            })
122            .collect()
123    }
124
125    pub(crate) fn find(&self, name: &str) -> Option<&dyn ErasedStep> {
126        self.definitions
127            .iter()
128            .find(|definition| definition.name == name)
129            .map(|definition| definition.handler)
130    }
131}
132
133impl<StartInput: Value> Default for StepList<'_, StartInput> {
134    fn default() -> Self {
135        Self::empty()
136    }
137}
138
139/// Targets a Step with encoded input and an optional per-movement options override.
140///
141/// Create movements with [`Self::to`] or [`Self::to_with_options`], then pass them to
142/// [`StepDecision::go_to_many`], [`crate::RpcResult::then`], or conditional completion.
143pub struct StepMovement {
144    pub(crate) target_step_type: &'static str,
145    input: Box<dyn ErasedValue>,
146    pub(crate) options_override: Option<ErasedStepOptions>,
147}
148
149impl StepMovement {
150    /// Targets `step` with typed `input` and the Step's registered options.
151    pub fn to<TargetStep>(step: &TargetStep, input: TargetStep::Input) -> Self
152    where
153        TargetStep: Step,
154    {
155        Self {
156            target_step_type: step.step_type(),
157            input: Box::new(TypedValue(input)),
158            options_override: None,
159        }
160    }
161
162    /// Targets `step` with typed input and options used only for this movement.
163    pub fn to_with_options<TargetStep>(
164        step: &TargetStep,
165        input: TargetStep::Input,
166        options: StepOptions<TargetStep::Input>,
167    ) -> Self
168    where
169        TargetStep: Step,
170    {
171        Self {
172            target_step_type: step.step_type(),
173            input: Box::new(TypedValue(input)),
174            options_override: Some(options.into()),
175        }
176    }
177
178    pub(crate) fn encode_input(&self) -> SdkResult<ProtoValue> {
179        self.input.encode()
180    }
181}
182
183/// Describes the durable outcome of one successful Step execution.
184///
185/// A decision moves to one or more Steps, completes or fails the Flow, waits for Channels to empty,
186/// or deliberately leaves the Flow at a dead end.
187pub struct StepDecision {
188    pub(crate) kind: StepDecisionKind,
189    pub(crate) cancel_step_types: Vec<&'static str>,
190    pub(crate) cancel_sibling_step_types: Vec<&'static str>,
191}
192
193pub(crate) enum StepDecisionKind {
194    Next(Vec<StepMovement>),
195    GracefulComplete(Box<dyn ErasedValue>),
196    ForceComplete(Box<dyn ErasedValue>),
197    ForceCompleteIfChannelsEmpty {
198        output: Box<dyn ErasedValue>,
199        fallback: Box<StepMovement>,
200        channels: Vec<ChannelGuard>,
201    },
202    ForceFail(String),
203    DeadEnd,
204}
205
206impl StepDecision {
207    /// Moves to one typed target Step.
208    pub fn go_to<TargetStep>(step: &TargetStep, input: TargetStep::Input) -> Self
209    where
210        TargetStep: Step,
211    {
212        Self::go_to_many([StepMovement::to(step, input)])
213    }
214
215    /// Moves to all targets, enabling concurrent active Steps.
216    pub fn go_to_many(movements: impl IntoIterator<Item = StepMovement>) -> Self {
217        Self {
218            kind: StepDecisionKind::Next(movements.into_iter().collect()),
219            cancel_step_types: Vec::new(),
220            cancel_sibling_step_types: Vec::new(),
221        }
222    }
223
224    /// Completes only after all other active Steps finish, returning `output`.
225    pub fn graceful_complete<Output: Value>(output: Output) -> Self {
226        Self {
227            kind: StepDecisionKind::GracefulComplete(Box::new(TypedValue(output))),
228            cancel_step_types: Vec::new(),
229            cancel_sibling_step_types: Vec::new(),
230        }
231    }
232
233    /// Completes immediately, abandoning other active Steps and returning `output`.
234    pub fn force_complete<Output: Value>(output: Output) -> Self {
235        Self {
236            kind: StepDecisionKind::ForceComplete(Box::new(TypedValue(output))),
237            cancel_step_types: Vec::new(),
238            cancel_sibling_step_types: Vec::new(),
239        }
240    }
241
242    /// Completes with `output` when every guard is empty; otherwise follows `fallback`.
243    pub fn force_complete_if_channels_empty<Output: Value>(
244        output: Output,
245        fallback: StepMovement,
246        channels: impl IntoIterator<Item = ChannelGuard>,
247    ) -> Self {
248        Self {
249            kind: StepDecisionKind::ForceCompleteIfChannelsEmpty {
250                output: Box::new(TypedValue(output)),
251                fallback: Box::new(fallback),
252                channels: channels.into_iter().collect(),
253            },
254            cancel_step_types: Vec::new(),
255            cancel_sibling_step_types: Vec::new(),
256        }
257    }
258
259    /// Fails the Flow immediately with an application-provided reason.
260    pub fn force_fail(reason: impl Into<String>) -> Self {
261        Self {
262            kind: StepDecisionKind::ForceFail(reason.into()),
263            cancel_step_types: Vec::new(),
264            cancel_sibling_step_types: Vec::new(),
265        }
266    }
267
268    /// Leaves the Flow running with no next Step.
269    ///
270    /// Use this only when an RPC or external operation will resume the Flow later.
271    pub fn dead_end() -> Self {
272        Self {
273            kind: StepDecisionKind::DeadEnd,
274            cancel_step_types: Vec::new(),
275            cancel_sibling_step_types: Vec::new(),
276        }
277    }
278
279    /// Selects every queued or active execution of one registered Step type.
280    ///
281    /// Dex resolves one snapshot after the current `execute` succeeds. Finished,
282    /// already-canceled, and absent executions are no-ops. Steps scheduled by this decision
283    /// are excluded. Call repeatedly to select multiple types; Flow-wide selection wins.
284    pub fn cancel_step<SelectedStep>(mut self, step: &SelectedStep) -> Self
285    where
286        SelectedStep: Step,
287    {
288        let step_type = step.step_type();
289        if !self.cancel_step_types.contains(&step_type) {
290            self.cancel_step_types.push(step_type);
291        }
292        self.cancel_sibling_step_types
293            .retain(|selected| *selected != step_type);
294        self
295    }
296
297    /// Selects same-source executions of one registered Step type.
298    ///
299    /// A sibling has the same [`Context::from_step_execution_id`](crate::Context::from_step_execution_id)
300    /// as the execution returning this decision. Snapshot and no-op behavior match
301    /// [`Self::cancel_step`]. Call repeatedly to select multiple Step types.
302    pub fn cancel_sibling_step<SelectedStep>(mut self, step: &SelectedStep) -> Self
303    where
304        SelectedStep: Step,
305    {
306        let step_type = step.step_type();
307        if !self.cancel_step_types.contains(&step_type)
308            && !self.cancel_sibling_step_types.contains(&step_type)
309        {
310            self.cancel_sibling_step_types.push(step_type);
311        }
312        self
313    }
314}
315
316#[derive(Clone)]
317pub(crate) struct RegisteredStep {
318    pub(crate) name: &'static str,
319    pub(crate) starting: bool,
320}
321
322struct StepDefinition<'a> {
323    name: &'static str,
324    starting: bool,
325    handler: &'a dyn ErasedStep,
326}
327
328impl<'a> StepDefinition<'a> {
329    fn new<SomeStep>(step: &'a SomeStep, starting: bool) -> Self
330    where
331        SomeStep: Step,
332    {
333        let name = step.step_type();
334        Self {
335            name,
336            starting,
337            handler: step,
338        }
339    }
340}
341
342pub(crate) trait ErasedStep: Send + Sync {
343    fn wait_for(&self, context: &mut Context, input: &ProtoValue) -> HandlerResult<Wait>;
344    fn execute(&self, context: &mut Context, input: &ProtoValue) -> HandlerResult<StepDecision>;
345    fn options(&self) -> ErasedStepOptions;
346}
347
348impl<SomeStep> ErasedStep for SomeStep
349where
350    SomeStep: Step,
351{
352    fn wait_for(&self, context: &mut Context, input: &ProtoValue) -> HandlerResult<Wait> {
353        Step::wait_for(self, context, value_mapper::decode_handler(input)?)
354    }
355
356    fn execute(&self, context: &mut Context, input: &ProtoValue) -> HandlerResult<StepDecision> {
357        Step::execute(self, context, value_mapper::decode_handler(input)?)
358    }
359
360    fn options(&self) -> ErasedStepOptions {
361        Step::options(self).into()
362    }
363}
364
365pub(crate) trait ErasedValue: Send + Sync {
366    fn encode(&self) -> SdkResult<ProtoValue>;
367}
368
369pub(crate) struct TypedValue<T>(pub(crate) T);
370
371impl<T: Value> ErasedValue for TypedValue<T> {
372    fn encode(&self) -> SdkResult<ProtoValue> {
373        value_mapper::encode(&self.0)
374    }
375}
dex_sdk/step.rs

dex_sdk/
step.rs
