Skip to main content

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.

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 loaded for its first attempt, so it does not see Attribute writes made after that load. 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.

ASYNC durability does this asynchronously. Dex Server starts the next work before the Step completion is persisted, and it batches these commits within about 20 milliseconds. It provides much higher throughput because it removes a lot of overhead from storage and the server.

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. Make ASYNC methods and the work they start idempotent.

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

Durability changes only when Dex Server persists a result. Retry policy, method timeout, and heartbeat settings apply to every attempt in both modes. The default method attempt timeout is two hours, and the default heartbeat timeout is one minute.

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 an 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.

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 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.

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. 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. 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(OverrideSecond, output, options))

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