Graceful Timeout
Use Dex's built-in graceful timeout when an overall Flow deadline needs application-defined handling. The timeout belongs to the Flow execution, not to a competing Step.
FlowGracefulTimeout starts LongWaitStep. When the Flow deadline expires, Dex runs handleTimeout. The handler response uses Execute semantics.
Core implementation
The complete Flow definition registers LongWaitStep and the timeout hook. Start the Flow with a positive timeout; HANDLER is explicit below and is also the default when the Flow provides the hook. The example gives the hook its own attempt timeout and retry policy with FlowTimeoutHandlerOptions.
class LongWaitStep(Step[bool]):
def wait_for(self, context: Context, input: bool) -> Wait:
if input:
return Wait.skip_immediately()
return Wait.until(Timer.by_duration(SLOW_TASK_DURATION))
def execute(self, context: Context, input: bool) -> StepDecision:
return force_complete("Workflow completed successfully")
class FlowGracefulTimeout(Flow[bool]):
def __init__(self) -> None:
self.long_wait_step = LongWaitStep()
def get_steps(self) -> StepList[bool]:
return StepList.start_step(self.long_wait_step)
def handle_timeout(self, context: Context) -> StepDecision:
return force_fail("Workflow did not finish the task in time")
await app_state.client.start_flow(
app_state.timeout,
flow_id,
optional_query("successfulWorkflow", "true").lower() == "true",
StartFlowOptions(
timeout=timedelta(minutes=1),
timeout_policy=FlowTimeoutPolicy.HANDLER,
timeout_handler_options=FlowTimeoutHandlerOptions(
method_timeout=timedelta(seconds=30),
retry=RetryPolicy(maximum_attempts=3),
),
),
)
Example: examples/python/dex_examples/patterns/timeout/flow_graceful_timeout.py
If the timeout handler succeeds, Dex applies its staged writes, best-effort Channel deletions, Channel publications, and decision. If it exhausts the configured retries, Dex fails the Flow.