Step options

This page covers the advanced StepOptions features to deeply customize your Step type. See the Basic Step page for the basic features.

Retry After

Use RetryAfter if you want to deeply customize the retry behavior. It overrides the retry schedule of the RetryPolicy.

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.

ASYNC durability does this asynchronously. Every few seconds, Dex Server flushes accumulated Step completions and persists them. It provides much higher throughput because it removes a lot of overhead from the backend database and server.

SYNC is the default. Dex resolves durability in this order: a method override in StepOptions, the Flow's FlowConfig, then SYNC.

SYNC runs only as a regular Temporal activity. Its default method attempt timeout is two hours, and its default heartbeat timeout is one minute.

ASYNC first runs as a Temporal local activity for at most seven seconds and three attempts, then falls back to a regular activity if it still fails. A smaller user retry duration or attempt limit still applies. The local activity phase ignores the method timeout and heartbeat settings. If it falls back to a regular activity, Dex applies the remaining retry budget and the same two-hour method attempt and one-minute heartbeat defaults.

The default total retry duration is four hours for both durability modes. An explicit retry duration may be shorter or longer.

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

Heartbeat

During a regular activity attempt, the Worker must send a heartbeat or a Stream message before the HeartbeatTimeout expires. In other words, sending a Stream message is an implicit heartbeat.

HeartbeatTimeout defaults to one minute. The default allowed minimum is 10 seconds.

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.

A heartbeat can persist a checkpoint value. The next regular attempt receives the latest 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.

The local phase of ASYNC durability ignores heartbeat output and does not pass a checkpoint to the fallback regular activity. Stream messages are still emitted during the local phase.

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

A heartbeat helps deliver cancellation. A Step invocation is canceled only when it completes, fails, or records a heartbeat.

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. Retries of one logical method call reuse its first 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 existing not-loaded error. A selected empty Channel returns an empty message list. Size and map-key introspection remains available without loading message bodies.

Attribute locks

Concurrent Steps and RPC invocations can read and write the same Attribute. This can lead to race conditions. To avoid this, 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 locking for more details.

Provide StepOptions Dynamically

The StepOptions returned by a Step implementation apply to every Step execution by default. Sometimes it is useful to override them dynamically for a certain 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(OverrideSecondStep, output, options))

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