Skip to main content

Step Basic Features

Step is the basic building block of background execution. This page covers the default shape of a Step: typed input, optional WaitFor, required Execute, movement with StepDecision, and the default StepOptions behavior.

A Dex Step is implemented by native code. The struct or class that implements the Step determines the StepType. In most languages, the SDK derives the StepType from the native type name unless you override it.

The same Step type can execute multiple times in one Flow. Each execution is a StepExecution, identified by a StepExecutionID in the form StepType-Number. Dex assigns and maintains that identifier.

Step input​

A step usually has an input that is uniquely persisted for this step execution.

Since a step can run multiple times and even in parallel, step input is very useful to tell the differences of different StepExecutions of the same step type.

The start Step of a flow is also the input of the flow.

For most languages, step input is strongly typed, when writing code to start a flow, or go to next step(s),

WaitFor and Execute​

A Step implementation has one or two small methods: WaitFor -> Execute.

WaitFor is optional. It returns a durable Wait to Dex Server. Dex persists that waiting state, evaluates the conditions on your behalf, and invokes Execute after the wait is satisfied.

If a Step does not implement WaitFor, Execute runs immediately.

Execute is required. It returns a StepDecision that moves to the next Step, branches to several Steps, or closes the Flow.

Each WaitFor or Execute invocation is a separate commit boundary. Dex Server stages the Attribute writes and Channel publications from that method, then commits them with its Wait or StepDecision only after the method returns successfully. If the method fails, none of those durable changes are committed. WaitFor and Execute do not share a commit. External API calls are outside this atomic boundary.

Before its final Wait or StepDecision, a handler can send heartbeat checkpoints and any number of best-effort Stream messages. The response stream must end with exactly one final result. These progress frames are not part of the method's commit, and a Stream Store failure does not fail the Step.

approval = Channel("Approval", str)


class StepSecond(Step[int]):
    def execute(self, context: Context, input: int) -> StepDecision:
        return graceful_complete(input + 1)


class ExampleStep(Step[int]):
    def __init__(self, second: StepSecond) -> None:
        self.second = second

    def wait_for(self, context: Context, input: int) -> Wait:
        return Wait.until(approval.for_one())

    def execute(self, context: Context, input: int) -> StepDecision:
        return go_to(StepSecond, input + 1)

Example: examples/python/dex_examples/primitives/step/step_flow.py

Durable WaitFor​

WaitFor returns Waiting to server using WaitingType and the conditions to wait for. Server persist the waiting state and evaluate the condition on any related changes happened. So the waiting is durable.

There are three waiting types: AnyOf,AllOf,AnyCombinationOf.

AnyOf and AllOf are the most frequently used/needed. See AnyCombinationOf as advanced feature.

As the name says,

  • the waiting condition is met when anyOf conditions is met.
  • the waiting condition is met when AllOf conditions are met.

Timer, Channel, and SubFlow are the conditions that the waiting type can wait on.

Until, AnyOf, and AllOf accept Conditions without condition IDs. Do not add IDs merely because a Condition is nested in one of these waits. AnyCombinationOf is the exception: every referenced Condition needs a unique condition ID.

return Wait.any_of(
    channel_a.for_one(),
    Timer.by_duration(timeout),
)

Example: examples/python/dex_examples/primitives/wait_types/wait_types_flow.py

return Wait.all_of(
    channel_a.for_one(),
    channel_b.for_one(),
)

Example: examples/python/dex_examples/primitives/wait_types/wait_types_flow.py

StepDecision​

A StepDecision can go to a single or multiple next steps.

This is how you build branching, looping, and parallel work.

See more in the Parallel Steps and Parallel SubFlows patterns.

Step Decision can also complete or fail the flow:

  • GracefulComplete will complete the flow when there are no more steps active in the flow
  • ForceComplete will complete the flow immediately
  • Fail will fail the flow immediately

And they all can return flow output so that client can wait and retrieve as FlowResult.

See more advanced Step decision for DeadEnd, Cancellation, forceCompleteIfChannelsEmpty.

StepOptions​

StepOptions is extremely powerful to customize the behavior of a step.

Most of the times, you need to customize failing handling -- timeout, retry and failure policy. There are more advanced Step options like heartbeat, durability etc.

  • Timeout customize the timeout of a single attempt of the method(WaitFor/Execute) invocation.
  • Retry policy — re-invokes the same method on the same Step execution until attempts are exhausted.
  • Failure policy — decide what happens after retries end. Default is to fail the Flow. But WaitFor can run Execute anyway, or Execute can go to a recovery Step.

Configure by implementing the step method to return a StepOptions.

def get_options(self) -> StepOptions:
    return StepOptions(
        execute_retry=RetryPolicy(
            initial_interval=timedelta(seconds=1),
            backoff_coefficient=2,
            maximum_attempts=5,
        ),
    )

def execute(self, context, ready_after_attempt: int) -> StepDecision:
    if context.attempt < ready_after_attempt:
        raise RuntimeError(f"not ready on attempt {context.attempt}")
    return graceful_complete("ready")

Example: examples/python/dex_examples/primitives/step/retry_flow.py

WaitFor failure policy​

When WaitFor method retries are exhausted:

  • FailFlow — Fail the flow. This is the default.
  • PROCEED — Continue to invoke Execute.

With PROCEED, the same Step's Execute receives the durable input unchanged. waitForMethodFailed() returns true. getRecoveryError() has the error_type and detail from the final failed WaitFor attempt.

def get_step_options(self) -> StepOptions:
    return StepOptions(
        wait_for_retry=RetryPolicy(maximum_attempts=2),
        wait_for_failure=WaitForFailurePolicy.PROCEED,
    )

def wait_for(self, context: Context, input: str) -> Wait:
    raise RuntimeError("planned WaitFor failure")

def execute(self, context: Context, input: str) -> StepDecision:
    if not context.wait_for_method_failed():
        raise RuntimeError("waitFor failure was not reported")
    return go_to(FinishStep, f"{input}_recovered")

Example: examples/python/dex_examples/primitives/proceed_on_wait_failure/proceed_on_wait_failure_flow.py

Execute failure policy​

When Execute method retries are exhausted, onExecuteFailureProceedTo / ExecuteFailure.proceedTo(...) proceed to a recovery Step with the same input.

def get_step_options(self) -> StepOptions:
    return StepOptions(
        execute_retry=RetryPolicy(maximum_attempts=5)
    ).on_execute_failure_proceed_to(UpdateQuantityRecovery)

Example: examples/python/dex_examples/patterns/recovery/failure_recovery_flow.py

See the Failure Handling pattern and money-transfer use case.

The recovery Step receives the failed Step's durable input unchanged. getRecoveryError() has the error_type and detail from the final failed Execute attempt.

Step Failed events and last failure info​

  • Terminal events StepExecuteFailed / StepWaitForFailed carry failure output when the Step ultimately fails.
  • During retries (before failure policy runs), Step events and live Flow state expose lastFailureInfo — the most recent attempt's error (type, detail, worker metadata).
  • In Dex Web, open event details to see Last failure while retries are in flight. Active Steps in the Flow overview also show lastFailureInfo.

Last failure info describes retry attempts. Failure policy decides what happens once retries end.