Skip to main content
Super Durable
English
English简体中文
GitHubLog inSign up Cloud
Menu

Flow and Step runtime

View as Markdown

This page holds the detailed runtime rules for Flows and Steps: start options, Flow configuration, waits, Step decisions, Step options, and what happens when an attempt fails or is interrupted. For the concepts, read Flow and Step first. For the exact signature of one API in your language, read its SDK doc comment.

Flow definition​

A Flow registers its Steps, its persistence schema, and its RPC handlers.

  • GetSteps / get_steps / steps registers one start Step and every other Step type the Flow may reach through Step decision movements. The start Step input type is also the Flow start input type. An empty Step list, or a list with no start Step, is valid when the Flow is driven entirely by RPCs that later move into Steps.
  • The persistence schema lists every Attribute and Channel used from Steps or RPCs. Dex rejects reads and writes to names that are not declared.
  • Each Flow explicitly declares its RPC handlers and their execution options. Go returns direct bound methods from GetRPCs with DefineRPC. Dex uses the method name as the durable RPC name, except Rust may provide an explicit name.

Most SDKs derive the FlowType and the StepType from the native type name unless you override it.

StartFlowOptions​

StartFlowOptions apply when a client starts a Flow or a SubFlow. They are immutable after start.

FieldPurpose
TimeoutFlow timeout. Omit it or set zero to disable it.
TimeoutPolicyWhat happens when the Flow timeout fires.
TimeoutHandlerOptionsExecution, retry, recovery, durability, locking, and state loading for a timeout handler.
IDReusePolicyHow to reuse an existing Flow ID to start.
RequestIDIdempotency key for retrying one start request.
AlreadyStartedOptionsHow to handle an already-started Flow.
RetryPolicyNot supported. server-zd rejects it with INVALID_ARGUMENT. See RetryPolicy.
StartDelayNot supported. server-zd rejects a positive value with INVALID_ARGUMENT. See StartDelay.
AttributesInitial Attribute values.
FlowConfigPer-execution Flow configuration, sent as ConfigOverride.

For an application-defined outcome at the Flow deadline, see Graceful Timeout.

def example_start_flow_options() -> StartFlowOptions:
    return StartFlowOptions(
        timeout=timedelta(minutes=30),
        timeout_policy=FlowTimeoutPolicy.HANDLER,
        id_reuse_policy=IdReusePolicy.DISALLOW,
        config_override=FlowConfig(step_durability=StepDurability.SYNC),
        ignore_already_started=True,
        request_id="start-order-123",
    ).with_attribute(status, "queued")

Example: examples/python/dex_examples/primitives/flow/controller.py

Timeout and TimeoutPolicy​

Timeout is one durable timer for the whole Flow. Set TimeoutPolicy only when Timeout is positive.

  • FAIL: fail the Flow with a timeout error.
  • CANCEL: cancel the Flow. It does not retry.
  • HANDLER: run one logical Flow timeout-handler execution.

With a positive Timeout, a Dex SDK uses FAIL by default. If the Flow has a timeout handler, it uses HANDLER instead. HANDLER is invalid when the Flow has no handler.

With a Timeout of 0, the Flow has no timeout. A Step that no Worker takes then waits forever. A SubFlow without a timeout behaves the same way. With a positive Timeout, FAIL and CANCEL end the wait when the timeout fires.

TimeoutHandlerOptions is valid only with a positive Timeout and a final HANDLER policy. It uses Execute semantics: a method timeout limits each attempt, Retry can run multiple attempts, HeartbeatTimeout detects stalled attempts, and Failure can route exhausted retries to a registered no-input Step. The recovery Step receives null or unit input and reads the final error from the handler Context.

A HANDLER timeout does not end queueing. The timeout handler Step also waits in the queue when no Worker takes it. Dex tries to dispatch it before other queued Steps, but it does not guarantee that order. Like any Step, the handler's method timeout starts when Dex sends an attempt to a Worker. Its TotalDuration starts when Dex first hands attempt 1 to a Worker. Its attempts may run past the original Flow deadline. If no failure target is configured, an exhausted handler fails the Flow.

Ordinary Attributes and all Channel size metadata load automatically. AttributeMap values and pending Channel messages require explicit selections in TimeoutHandlerOptions. The handler may stage Attribute writes, Channel deletions, and Channel publications before returning its decision. Flow-level retry does not exist, so the timeout handler runs once per Flow run.

Set TimeoutPolicy to HANDLER to run this code when the timeout fires. This handler records why the Flow ended, then fails it with an application reason.

def handle_timeout(self, context: Context) -> StepDecision:
    status.set(context, "timed out")
    return force_fail("processing deadline reached")

Example: examples/python/dex_examples/primitives/flow/example_flow.py

Flow ID, RequestID, and AlreadyStartedOptions​

IDReusePolicy says what to do when this Flow ID already exists. RequestID says whether this call retries the same start request.

These options answer different questions, in this order:

  1. RequestID identifies one start request. When omitted, the SDK generates a new UUID, so another StartFlow call is a new request. Reuse it only for a retry of the exact same start. Dex compares the RequestID, not the start input or options.
  2. IDReusePolicy decides whether that request may start another run with this Flow ID.
  3. AlreadyStartedOptions.IgnoreError matters only when the ID reuse policy rejects the start as already started. It returns the existing Run ID only when the RequestID matches.
Retry requestAlreadyStartedOptions.IgnoreErrorWhen the ID reuse policy rejects the start
No RequestIDOff or onThe SDK sends a new UUID. Dex returns an already-started error.
A different RequestIDOff or onDex returns an already-started error.
Same RequestIDOffDex returns an already-started error.
Same RequestIDOnDex returns the existing Run ID.

The ID reuse policy determines whether Dex rejects the start in the first place:

IDReusePolicyExisting active FlowExisting closed Flow
IDReuseDefaultRejects the start. A matching RequestID with IgnoreError returns the active Run ID.Rejects the start. A matching RequestID with IgnoreError returns the existing Run ID.
IDReuseAllowIfPreviousFailedRejects the start. A matching RequestID with IgnoreError returns the active Run ID.Starts a new run only after a failed run; otherwise rejects it.
IDReuseAllowIfNotRunningRejects the start. A matching RequestID with IgnoreError returns the active Run ID.Starts a new run.
IDReuseDisallowRejects the start. A matching RequestID with IgnoreError returns the existing Run ID.Rejects the start. A matching RequestID with IgnoreError returns the existing Run ID.
IDReuseTerminateIfRunningTerminates the active run and starts a replacement. It does not produce an already-started error.Starts a new run.

What Dex Server does​

Dex Server requires a non-empty RequestID. An SDK generates one before sending the request when you omit it. The server records it with the Flow execution, then starts the Flow using the requested ID reuse policy. IDReuseDefault uses the server default: reject every previously used Flow ID. Choose IDReuseAllowIfNotRunning explicitly when a closed Flow ID should start a new execution.

If the Flow is already started, Dex checks AlreadyStartedOptions.IgnoreError. When it is on, Dex reads the existing run's recorded RequestID. It returns that Run ID only when the two RequestIDs match. Otherwise it returns the already-started error.

StartFlow best practices​

Call StartFlow directly with a stable business Flow ID. Do not read a snapshot, search for the Flow, or check its status first just to avoid a possible retry conflict. A successful start also needs no read to recheck the identity, owner, or initial Attributes you supplied.

For retries of one logical start, combine a stable RequestID, IDReuseDisallow, and the ignore-already-started option. Keep the same input and options on every retry. IDReuseDisallow prevents another execution after closure; a policy that permits another run can bypass the matching-RequestID path.

The following helpers accept a validated integer input and a stable Flow ID. Those are the complete business inputs in this example, so together they determine the RequestID. All other start values are fixed. In a real application, include every logical input that changes the operation, such as tenant, owner, mode, revision, or initial business Attributes. Dex compares RequestIDs; it does not compare payload equality. Do not reuse a RequestID for a changed logical request.

Each helper returns the known business Flow ID after acceptance and propagates errors. It performs no read before or after the start. The returned ID is an acknowledgement, not a current business snapshot or proof that the Flow has completed.

async def start_example_flow_request(
    client: AsyncClient, example_flow: ExampleFlow, flow_id: str, input_num: int
) -> str:
    options = StartFlowOptions(
        id_reuse_policy=IdReusePolicy.DISALLOW,
        request_id=f"start-example-flow:{input_num}:{flow_id}",
        ignore_already_started=True,
    ).with_attribute(status, "queued")
    await client.start_flow(example_flow, flow_id, input_num, options)
    return flow_id

Example: examples/python/dex_examples/primitives/flow/controller.py

For example, starting invoice-42 with input 7 sends RequestID start-example-flow:7:invoice-42. Retrying that call returns the same execution, even after it closes, while Dex retains it. Starting the same Flow ID with input 8 sends a different RequestID and returns an already-started error. No snapshot comparison is needed on the successful path.

Wait when the response promises a business milestone​

If the API must confirm an important operation, call WaitForAttributeMatch or WaitForStepCompletion immediately after StartFlow succeeds, including matching-request deduplication. Deduplication confirms the same start was accepted; it does not prove the operation has finished. No identity-check read is needed between the start and the wait.

For example, an API may promise that a row has been committed to a database table used as the business source of truth. The owning Flow writes that row in Execute, then records a committed Attribute or completes the persistence Step. The API waits for that committed Attribute value or the specific StepExecutionID before returning confirmed persistence. Publish the marker only after the database commit succeeds. WaitForStepCompletion confirms that Step completed; it does not return its output or wait for the whole Flow to finish.

API → StartFlow succeeds or deduplicates → wait for the required commit milestone
Flow Execute → commit business row to database → record committed Attribute / complete Step
API → milestone wait succeeds → return confirmed persistence

Wait only for the condition that establishes the response's promised outcome. If the response only acknowledges accepted background work, return immediately after the start without a wait. Bound required waits with a caller deadline or request timeout. A wait timeout does not undo the start or prove the database write failed; report completion as pending or unknown according to the API contract and preserve the same start identity for retries.

See Wait for a Step and Wait for an Attribute match for these waits. Attribute waits target an active run. If a Flow closes before the wait attaches, reconcile the actual wait error through the authoritative business record or a supported terminal read; do not assume the Attribute wait can query a closed run.

Read only when the error outcome needs it​

Error handling follows the business contract. If an already-started error can simply become ALREADY_EXISTS, return that conflict directly. Do not add a read-only RPC just because the error occurred. The helpers above preserve the typed error for that boundary translation.

Read existing state only when it changes the response or next business decision. For example, an API that explicitly permits continuing an existing operation may call its typed read-only RPC after the conflict to decide whether that operation satisfies the request. An API that promises a current snapshot must read current state; it cannot label the submitted initial Attributes as that snapshot. An accepted start does not prove business completion.

SituationAction
Start accepted; response only acknowledges acceptanceReturn the Flow ID and the agreed acceptance fields. No snapshot read.
Start accepted or deduplicated; response promises a critical operationWait for the committed Attribute condition or selected Step completion, then return that milestone outcome.
Already started; business accepts ALREADY_EXISTSReturn the conflict. No snapshot read.
Already started; business permits resuming an existing operationRead its typed snapshot only to decide whether that contract permits resuming it.
Response lost; start acceptance is unknownRetry within a bounded policy with the same identity and options, or read authoritative state if it can establish the outcome.

Do not retry a proven conflict indefinitely. If no stable RequestID was supplied, separate SDK calls use fresh IDs; enabling the ignore option alone does not deduplicate them.

Put dependent starts inside the owning Flow​

For one business operation, the API starts one owning Flow. Its durable Steps perform the follow-on work. Do not start the first Flow, read its status, and start a dependent Flow in the HTTP handler. If that process exits between the starts, no durable Step owns the missing launch.

Use Steps when the work shares one identity and lifecycle. When it needs a separate top-level Flow with its own owner and lifecycle, launch it from an Execute Step with stable downstream Flow ID, RequestID, and start options. Persist the launch input in the owning Flow. If the Worker dies after the downstream start is accepted but before the Step commits, the Step can retry the same start without creating another execution. Keep WaitFor for durable conditions. A separately owned top-level Flow does not need to become a SubFlow merely because another Flow starts it.

API → StartFlow for the owning operation → accepted response
Owning Flow Execute → StartFlow for separately owned work → durable progress
Worker replacement → retry the same downstream start identity → no duplicate launch

RetryPolicy​

server-zd does not support a Flow-level RetryPolicy. A start request that sets one fails with INVALID_ARGUMENT. Tracked in issue 996. To retry work, use Step retry, which is supported.

StartDelay​

server-zd does not support StartDelay. A start request with a positive delay fails with INVALID_ARGUMENT. Tracked in issue 995. To delay the first work, start the Flow without a delay and begin the start Step with a durable Timer wait.

Attributes​

Attributes write initial values. Each value must match an Attribute or AttributeMap in the Flow persistence schema.

FlowConfig​

FlowConfig holds runtime behavior. Unlike StartFlowOptions, it is mutable after start.

Use ConfigOverride in StartFlowOptions to change the initial configuration for one Flow. It changes only the fields you provide. It does not replace the whole FlowConfig. At start, Dex applies the supplied fields over the Dex Server defaults. Omitted fields keep their default values.

FieldBehavior
ActiveStepSearchModeIndex every active Step, only Steps that ran WaitFor, or no active Steps. The built-in server default does not index active Steps. Set an explicit mode to opt in.
ContinueAsNewThresholdDeprecated. Dex does not continue Flows as new. Dex rejects a negative value, stores the value in the FlowConfig, and otherwise ignores it. ContinueAsNewPageSizeBytes behaves the same way.
StepDurabilityDefault durability for methods without a Step override. Dex resolves a method override first, then this value, then SYNC. Dex pins the durability when a method call becomes pending, so a change affects only method calls scheduled afterward.
AttributeStoreNamesDex Server Attribute Stores for Attributes that opt in to Attribute sync.

Update FlowConfig with a Client​

Call UpdateFlowConfig while the Flow is active. Send only the fields you want to change. Dex merges them with the existing configuration. The SDK UpdateFlowConfig call does not change the Flow's Worker routing.

Step methods and commits​

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.

Waits​

WaitFor returns a Wait with a wait type and the conditions to wait for. Dex persists the waiting state and evaluates the conditions when related state changes. Timer, Channel, and SubFlow are the condition kinds.

  • Until waits for one condition.
  • AnyOf is met when any condition is met.
  • AllOf is met when all conditions are met.
  • AnyCombinationOf is met when any one group of conditions is all met.

Until, AnyOf, and AllOf accept Conditions without condition IDs. Do not add IDs merely because a Condition is nested in one of these waits. Read Channel results from the Channel definition and inspect Timer outcomes through the Context. Add IDs only when another API requires them, such as AnyCombinationOf or a Timer selected by 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

Condition results​

After WaitFor completes, Execute can read what satisfied the Wait. For a Channel, use GetConditionResults / results / condition_results to read published messages. For a SubFlow, use SubFlowResult / getConditionResults / condition_result to decode the child Flow output.

Channel consumption is not greedy across AnyOf alternatives. Dex consumes messages only from the Channel condition selected to satisfy the Wait. Other Channel conditions consume nothing even if they are also ready, so their messages remain queued.

When several AnyOf alternatives are ready in one evaluation, Dex selects the first feasible candidate in canonical order: Timer conditions in declaration order, then Channel conditions in declaration order, then SubFlow conditions in declaration order. An earlier condition that is not ready does not block a later ready condition. Mixed condition kinds therefore do not share one global call argument order.

Use AnyOf for a race among the current alternatives. For strict priority that must keep waiting for a higher-priority input, return only that condition until it is resolved.

Channel results​

class ChannelWaitStep(Step[int]):
    def __init__(self, approval: Channel[str]) -> None:
        self.approval = approval

    def wait_for(self, context: Context, input: int) -> Wait:
        return Wait.any_of(
            self.approval.for_one(),
            Timer.by_duration(timedelta(seconds=input)),
        )

    def execute(self, context: Context, input: int) -> StepDecision:
        if context.has_timer_fired():
            return graceful_complete("approval timed out")
        approvals = self.approval.results(context)
        return graceful_complete(approvals[0])

Example: examples/python/dex_examples/primitives/channel/channel_flow.py

SubFlow results​

def execute(self, context: Context, input: int) -> StepDecision:
    result = SubFlow.get_condition_results(context)
    output = result.single_output(int)
    return graceful_complete(f"{SubFlow.get_flow_id(context)}|{output}")

Example: examples/python/dex_examples/primitives/subflow/parent_flow.py

AnyCombinationOf​

AnyCombinationOf expresses an OR of AND groups. Build each inner group as an all-of combination.

Channel consumption is not greedy across combinations. Dex consumes messages only from Channel conditions in the selected combination. Conditions belonging only to other ready combinations consume nothing, so their messages remain queued.

Every Condition in its combinations requires a unique condition ID.

return Wait.any_combination_of(
    ConditionCombination.of(
        channel_a.for_one(condition_id="signal-a"),
        Timer.by_duration(timeout, condition_id="timeout"),
    ),
    ConditionCombination.of(
        channel_b.for_one(condition_id="signal-b"),
    ),
)

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

Step execution local​

SetStepExecutionLocal stores a value on this StepExecution so Execute can read it with GetStepExecutionLocal. Use it when WaitFor computes something Execute needs but a Flow-level Attribute is overkill. Step input is already passed to both methods. Do not copy input into a local.

def wait_for(self, context: Context, input: int) -> Wait:
    context.set_step_execution_local("note", f"approval:{input}")
    return Wait.until(approval.for_one())

def execute(self, context: Context, input: int) -> StepDecision:
    note = context.get_step_execution_local("note", str)
    return graceful_complete(note or "")

Example: examples/python/dex_examples/primitives/step_execution_local/step_execution_local_flow.py

Skip immediately​

SkipWaitImmediately waits for no condition. Use it in a dynamic WaitFor method when one branch must not wait for anything.

return Wait.skip_immediately()

Example: examples/python/dex_examples/primitives/flow/example_flow.py

Step decisions​

Execute returns a StepDecision. It moves to next Steps, closes the Flow, or both stops this branch and cancels other Steps.

Multiple next Steps​

GoTo starts one next Step. GoToMany starts multiple next Steps. Each Step movement has its own input and can also have its own Step options.

return go_to_many(
    StepMovement.of(CarrierA, Quote(carrier="A", price=10)),
    StepMovement.of(CarrierB, Quote(carrier="B", price=12)),
    StepMovement.of(Winner, quote),
)

Example: examples/python/dex_examples/primitives/step_decision/step_decision_flow.py

Close decisions​

DecisionWhat it does
GracefulCompleteStops this branch and triggers the Flow to prepare for completion. When all branches and Steps are stopped, the Flow completes.
DeadEndStops this branch only.
ForceCompleteCompletes the Flow immediately.
ForceFailFails the Flow immediately.

GracefulComplete and ForceComplete can return an output. ForceFail returns a failure reason. The Flow result returns each output with the Step type and Step execution ID. When multiple Steps complete gracefully, all their outputs return together.

A Flow can have no active Steps and still keep running. DeadEnd stops this branch without making a decision that triggers the Flow to complete.

if mode == "graceful":
    return graceful_complete("done")
if mode == "dead-end":
    return go_to_many(
        StepMovement.of(BranchWorker, "left"),
        StepMovement.of(BranchWorker, "right"),
    )

return dead_end()

Example: examples/python/dex_examples/primitives/step_decision/step_decision_flow.py

def execute(self, context: Context, input: None) -> StepDecision:
    return force_fail("Workflow did not finish the task in time")

def execute(self, context: Context, input: bool) -> StepDecision:
    return force_complete("Workflow completed successfully")

Example: examples/python/dex_examples/patterns/timeout/flow_graceful_timeout.py

Cancel other Steps​

A StepDecision can cancel other active Steps.

  • CancelSteps selects active Steps by Step type.
  • CancelSiblingSteps selects active Steps that were triggered by the same parent Step execution.

An RPC can also cancel Steps by type. See Cancel running Steps. For how a running attempt observes cancellation, see Cancellation.

return (
    go_to(RecordQuote, quote)
    .with_canceling_steps(self.flow.carrier_a, self.flow.carrier_b)
)

Example: examples/python/dex_examples/primitives/step_decision/step_decision_flow.py

ForceCompleteIfChannelsEmpty​

ForceCompleteIfChannelsEmpty atomically checks whether all named Channels are empty. A Channel with queued messages is non-empty. After a message is consumed and committed to a Step, that Channel remains non-empty until the Step finishes processing Execute. If every named Channel is empty, it completes the Flow. Otherwise, it goes to other Steps.

Use it to drain Channels and complete the Flow, so the Flow can stay short. See Drain External Channel for the full example.

return force_complete_if_channels_empty(
    None,
    StepMovement.of(ProcessMessage, None),
    self.queue_channel,
)

Example: examples/python/dex_examples/patterns/drain-channels/external_publishing/draining_channel_flow.py

StepOptions​

A Step type returns StepOptions to customize its methods. WaitFor and Execute have separate retry policies, timeouts, failure policies, state selections, and lock sets.

Override StepOptions per Step execution​

The StepOptions returned by a Step implementation apply to every Step execution by default. To override them for one Step execution, provide the StepOptions override when a StepDecision goes to the next Steps.

options = StepOptions(
    wait_for_retry=RetryPolicy(maximum_attempts=2),
    wait_for_failure=WaitForFailurePolicy.PROCEED,
)
return go_to_many(StepMovement.of(OverrideSecond, output, options))

Example: examples/python/dex_examples/primitives/options_override/options_override_flow.py

Retry and failure policies​

  • Timeout limits a single attempt of the method invocation. See Timeouts and total duration.
  • Retry policy re-invokes the same method on the same Step execution until attempts are exhausted.
  • Failure policy decides what happens after retries end. The default fails the Flow. WaitFor can run Execute anyway, and Execute can go to a recovery Step.
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 retries are exhausted:

  • FailFlow fails the Flow. This is the default.
  • PROCEED continues 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(Finish, f"{input}_recovered")

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

Execute failure policy​

When Execute retries are exhausted, onExecuteFailureProceedTo / ExecuteFailure.proceedTo(...) proceeds 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

The recovery Step receives the failed Step's durable input unchanged. getRecoveryError() has the error type and detail from the final failed Execute attempt, and its attempt progress. See the Failure Handling pattern and the money-transfer use case.

Last failure info​

  • Terminal events StepExecuteFailed / StepWaitForFailed carry failure output when the Step ultimately fails.
  • During retries, before the failure policy runs, Step events and live Flow state expose lastFailureInfo: the most recent attempt's error type, detail, and 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.

RetryAfter​

Use RetryAfter to choose the delay before the next attempt. It overrides the retry schedule of the RetryPolicy.

Retries recover from failed attempts. Do not use RetryAfter to wait for an external system; use the Polling pattern.

A retry of one Execute call reuses the state snapshot that Dex took when it handed the first attempt to a Worker, so it does not see Attribute writes made after that. To wait for an external write, have the writer also publish a Channel message, directly or from an RPC handler, and wait on that message in the WaitFor of the next Step. Alternatively, the next Step can wait on a Timer and read fresh state when it runs.

from dex._grpc_errors import retry_after

if context.attempt < ready_after_attempt:
    raise retry_after(7, RuntimeError(f"not ready on attempt {context.attempt}"))

Example: examples/python/dex_examples/primitives/custom_retry/custom_retry_flow.py

Durability​

SYNC durability records the Step completion in Dex Server and persists it to storage synchronously. It also makes the attempt start durable in storage before the Worker receives the task.

ASYNC durability does this asynchronously. Dex Server hands the task to the Worker and starts the next work before the start and the completion are persisted, and it batches these commits within about 20 milliseconds. Sustained server throughput is about the same for both modes. ASYNC has lower latency, and a single Flow runs faster. On object storage with low write latency, ASYNC also makes far fewer requests.

If a Dex Server node fails before the commit, a recent ASYNC result is lost. The method runs again, and so does any later work that started from that result. If the node fails before the ASYNC start was persisted, Dex rolls the start back and runs the same attempt number again. This consumes no retry, and the attempt can see different input. Two executions of the same attempt number can overlap, and Step Stream writes can repeat in the same source. Make ASYNC methods and the work they start idempotent.

Choose SYNC for effects that need the tightest guarantees. Choose ASYNC for short Steps where latency matters and a rare rerun is acceptable. See Failure and redelivery for the full comparison.

SYNC is the default. Dex resolves durability in this order: a method override in StepOptions, the Flow's FlowConfig, then SYNC. Dex pins the durability when a method call becomes pending. Changing step_durability in the Flow config affects only method calls scheduled afterward.

Retry policy, method timeout, and heartbeat settings apply to every attempt in both modes.

def get_step_options(self) -> StepOptions:
    return StepOptions(execute_durability=StepDurability.SYNC)

def get_step_options(self) -> StepOptions:
    return StepOptions(execute_durability=StepDurability.ASYNC)

Example: examples/python/dex_examples/primitives/durability/durability_flow.py

Timeouts and total duration​

Three options bound the time a Step method can take.

LimitOptionDefaultEnforcement
Longest single attemptWaitForMethodTimeout or ExecuteMethodTimeoutTwo hours. The server has no hard cap.Dex times out the attempt. Another attempt starts only while TotalDuration has time left.
Time budget for retriesTotalDuration in the retry policyFour hours, for both durability modes.After an attempt fails, Dex retries only while the budget has time left. It never stops an attempt in progress.
Stalled WorkerHeartbeatTimeoutOne minute. The default allowed minimum is 10 seconds.Dex times out an attempt that sends no heartbeat or Stream message within this window. The retry policy then starts another attempt. See Heartbeat.

The method timeout and the heartbeat timeout start when Dex sends the task to the Worker. Waiting for locks or for a Worker, including time in the task queue, does not count. The deadline in the task is the send time plus the method timeout.

TotalDuration starts when Dex first hands attempt 1 to a Worker. If attempt 1 was returned without running, the clock resets. Dex does not check TotalDuration while a Step waits in the task queue, so a retry that waited past it still runs. A retry waits in the queue again. Dex dispatches queued tasks roughly in the order they became ready, with no strict order. That wait counts toward TotalDuration. After a long backlog, a failed retry may have no time left.

Dex uses TotalDuration only after an attempt fails, to decide whether to retry. It never cuts a running attempt, so the last attempt can start or end after TotalDuration. Only the method timeout limits a single attempt. To bound one long attempt, set ExecuteMethodTimeout or WaitForMethodTimeout, not TotalDuration.

To bound a long wait, set the method timeout and TotalDuration to the same maximum. Dex enforces both, so the Step does not need its own deadline. With Workers available, the total wait is at most about TotalDuration plus one method timeout. Give the retry policy a few attempts with backoff for transient errors and Worker crashes.

When a timeout expires, Dex fails the attempt with a timeout error. After retries end, the failure policy decides what happens next. If no failure route is configured, the Flow fails. See Execute Failure Recovery. For a long wait on an external system, see Polling.

Heartbeat​

During an attempt, the Worker must send a heartbeat or a Stream message before the HeartbeatTimeout expires. Sending a Stream message is an implicit heartbeat.

For example, an Execute method timeout of 60 seconds and a HeartbeatTimeout of 10 seconds lets Dex retry the Step in about 10 seconds instead of waiting for the full 60 seconds. To wait for an external system in one heartbeating Step, use the Polling pattern.

A heartbeat can persist a checkpoint value. The next attempt receives the latest stored value and can resume from it. An explicit heartbeat without a value clears the previous checkpoint. An implicit heartbeat from a Stream message preserves the latest explicit state, including an explicitly cleared state. Checkpoints are lossy; see Heartbeats are not acknowledged.

def get_step_options(self) -> StepOptions:
    return StepOptions(
        execute_method_timeout=timedelta(seconds=60),
        heartbeat_timeout=timedelta(seconds=10),
        execute_retry=RetryPolicy(maximum_attempts=3),
    )

async def execute(
    self, context: AsyncContext, batches: int
) -> StepDecision:
    completed_batches = context.get_last_heartbeat_value(int) or 0
    for batch in range(completed_batches, batches):
        if context.is_cancellation_requested():
            return dead_end()
        await asyncio.sleep(2)
        await context.heartbeat(batch + 1)
    return graceful_complete("processed")

Example: examples/python/dex_examples/primitives/heartbeat/heartbeat_flow.py

Cancellation​

Dex cancels a running attempt directly. It sends a cancel request to the Worker when the Step is canceled, for example by a StepDecision or an RPC, and when the attempt times out. The SDK also cancels the handler when the attempt deadline in the task passes. The handler does not need to complete, fail, or record a heartbeat to receive cancellation.

Cancellation is cooperative. The SDK cannot stop running code. A long handler must check for cancellation and return, as the heartbeat example above does. See Make the Step stop for the check in each language, and Cancel running Steps for how Dex ends a canceled attempt.

Because cancellation is cooperative, an interrupted execution may still run while the next attempt or the failure route starts.

Selective state loading​

Every WaitFor, Execute, and timeout-handler call receives all ordinary Attributes and size metadata for every Channel and ChannelMap instance. AttributeMap values and pending Channel messages are not loaded unless that method selects them.

StepOptions has separate selections for WaitFor and Execute. Select an entire AttributeMap or ChannelMap when the method must enumerate current instances. Select an exact map instance when its key is already known. An Attribute lock controls concurrency but does not load the AttributeMap value.

The two methods receive independent snapshots. Execute reads its snapshot after the winning Wait conditions consume their messages, so it does not see those consumed messages again. Dex takes the snapshot of attempt 1 when it hands the attempt to a Worker. Retries of one logical method call, and attempts that return to pending, reuse that snapshot.

This Step loads pending Queued messages only for Execute. It deletes the first pending message in the same successful handler response.

def get_step_options(self) -> StepOptions:
    return StepOptions(execute_load_channels=(self.queued,))

Example: examples/python/dex_examples/primitives/channel/channel_flow.py

Reading an unselected AttributeMap value or pending Channel message returns the SDK's not-loaded error. A selected empty Channel returns an empty message list. Size and map-key introspection remains available without loading message bodies. For Go, dexcli visualize reports a Step method that reads unselected state. See State loads.

Attribute locks​

Concurrent Steps and RPC invocations can read and write the same Attribute. To avoid races, use Attribute locks. Dex Server ensures only one Step execution or RPC handling can acquire the lock at a time. WaitFor and Execute have separate lock sets. Set them by returning StepOptions.

See Attribute locks for when a Step attempt takes and holds its locks, with WaitFor and Execute examples. See Locks for the lock model.

Failure and redelivery​

Retries and failure routes handle attempts that fail. Some failures leave Dex without knowing whether the work ran. A Step with an external effect that must not repeat, such as a payment or an email, has to plan for these cases. This section states what an application can rely on.

Heartbeats are not acknowledged​

A heartbeat call returns once the Worker has queued or sent it. It does not wait for Dex to store anything. Python's heartbeat, Go's RecordHeartbeat, Java's and TypeScript's recordHeartbeat, and Rust's record_heartbeat all behave this way. A returned heartbeat call does not prove that Dex stored the checkpoint. This holds for SYNC Steps too, even though a SYNC Step runs user code only after Dex made the attempt start durable.

Heartbeat checkpoints are lossy. Dex batches them and keeps only the latest value of each batch. A checkpoint can also be lost when the attempt ends, such as on a timeout or a cancellation, or when a Dex Server node fails. After an interruption, the next attempt receives the last checkpoint that Dex stored. That checkpoint can be older than the last one the Step sent. Do not use a checkpoint as the only record that an external effect happened.

An attempt starts when Dex hands it to a Worker​

A Step attempt starts at the moment Dex hands it to a Worker. Dex first reserves an idle Worker slot. It then records the start. This is the record that pins the attempt's input snapshot and takes its Attribute locks. Dex sends the task last. For a SYNC Step, the start is durable in object storage before the Worker receives the task. An ASYNC Step does not wait for that write.

Once Dex handed an attempt to a Worker, Dex never hands the same attempt number to another Worker. If the Worker crashes or its connection breaks, the attempt fails whether or not it sent output. The retry policy then runs: it consumes a retry and waits for the backoff. If a Worker stops responding without disconnecting, the attempt fails when its heartbeat timeout expires. A server restart or failover does the same for every attempt that was handed out. Those attempts fail with the failure kind STARTED_ATTEMPT_INTERRUPTED. SYNC and ASYNC behave the same in these cases.

A few cases return the attempt to pending instead. The attempt keeps its attempt number, its input snapshot, and its locks, and it consumes no retry. This happens only when the Worker certainly never ran the attempt:

  • The Worker declines the Step type as unknown. Dex then avoids that Worker for that Step type.
  • The task never reached the Worker, because the stream closed before the send or the send failed.
  • A Worker that uses reserved-slot dispatch closes its stream cleanly while the task has no output yet. This is a normal Worker shutdown. A shutting-down Worker finishes its running tasks, sends their results, and then closes. Tasks that arrive while it closes are not run and return to Dex.
  • An older SDK Worker declines the task for lack of capacity.

An ASYNC Step has one more case. If a Dex Server node crashes before Dex wrote the ASYNC start to object storage, Dex rolls that start back. The same attempt number then runs again, with no retry consumed and possibly on different input. So for ASYNC, one attempt number can run twice, and the two executions can overlap. Step Stream writes can then repeat in the same source.

For both durabilities, the attempts of one Step are not strictly sequential. Cancellation is cooperative, so an interrupted execution may still run while the next attempt or the failure route starts.

SYNC (default)ASYNC
Workers one attempt number reachesAt most one, except when the attempt returns to pending as described above.At most one, except a rollback after a node crash can run the same number again.
Start and execution orderThe start is durable before the Worker receives the task.The Worker can run before the start is durable.
Worker disconnect or node failoverThe attempt fails and the retry policy runs.The same. A node crash before the start was written is the rollback case.
attempt_progressNOT_STARTED or NOT_COMPLETED.Always NOT_COMPLETED.
Overlapping attemptsPossible after cancellation, because cancellation is cooperative.Possible after cancellation, and between two executions of the same attempt number.
Step Stream writesCan repeat across attempts.Can also repeat within one attempt number, in the same source.
When to useEffects that need the tightest guarantees.Short Steps where latency matters and a rare rerun is acceptable.

A failure route cannot tell whether the work ran​

A Step attempt that no Worker takes does not fail. It waits in the durable task queue, uses no retry, and runs when a Worker takes it. Only an RPC fails when no Worker takes it within the dispatch wait. A Step attempt that does fail can fail before or after a Worker received it, and its error type and detail alone do not always tell which.

The recovery error (RecoveryErrorInfo) also carries attempt_progress. It is the furthest progress across all attempts of the failed method call:

  • NOT_STARTED: no attempt ever reached a Worker that could have run it. The handler certainly did not run.
  • NOT_COMPLETED: at least one attempt was handed to a Worker and may have run, but none completed.
  • UNSPECIFIED: an older server did not report progress, or the call started before the upgrade. Treat it like NOT_COMPLETED.

ASYNC failures always report NOT_COMPLETED, because a rolled-back execution may have run.

NOT_COMPLETED is conservative. Dex can report it although the Worker never ran the handler:

  • The server crashed after the start was durable but before the task reached the Worker.
  • The connection broke right after Dex wrote the task but before the Worker read it.
  • The Worker declined the task or closed cleanly, but the server crashed before it recorded that.
  • A failure inside the SDK before it called the handler, such as reading input blobs or decoding. Dex reports this as a normal Worker error. A misconfigured FlowServiceAddress makes every attempt with blob inputs fail this way. values.hydrateBeforeDispatch avoids it.

A failure route can rely on NOT_STARTED to mean the work did not run. It cannot rely on NOT_COMPLETED to mean the work ran. The Flow's history records what Dex knows, and Dex Web shows it.

Design for these limits​

  • Make each external effect idempotent. Key it by an ID that stays the same across attempts, such as the Flow ID and the Step execution ID, or your own business key. Send the key to the service that applies the effect, so that it applies the effect once.
  • Treat a Step with a single attempt as at most once only within these limits. An interrupted attempt may have made its call, and a cancelled execution may still run. Its failure route can run although the work happened, or although it never ran. For ASYNC, the same attempt number can also run twice.
  • Reconcile an unknown outcome. When a failure route sees NOT_COMPLETED, check the target system by the idempotency key or your business key before you repeat or report the effect. Record the outcome you find in the Flow.

For the agent packages' single-attempt tool Steps, see Tool call limits.

Version notes​

Changed in 0.1.17 (Server protocol 4, required by SDK 1.35):

  • A Step attempt deadline is the hand-out time plus the method timeout. It is no longer shortened to the remaining retry budget. Queue and lock waits are excluded.
  • An interrupted attempt that was handed out consumes a retry.
  • TotalDuration never ends a running attempt. The server no longer produces the RETRY_DEADLINE failure kind. The SDK helper IsRetryTotalDurationExceeded, and its equivalent in other SDKs, can be true only for records written before the upgrade.