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

dex_sdk/
step.rs
