Graceful Timeout
当整个 Flow 的截止时间需要应用代码处理时，使用 Dex 内置的 graceful timeout。超时属于 Flow execution，而不是与业务 Step 竞争的另一个 Step。
FlowGracefulTimeout 启动 LongWaitStep。Flow 截止时间到达后，Dex 运行 handleTimeout。Handler response 使用 Execute 语义。
核心实现
完整的 Flow 定义会注册 LongWaitStep 和 timeout hook。使用正超时启动 Flow；以下示例显式选择 HANDLER，而当 Flow 提供 hook 时它也是默认策略。示例通过 FlowTimeoutHandlerOptions 为 hook 设置自己的 attempt timeout 和 retry policy。
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),
),
),
)
例子: examples/python/dex_examples/patterns/timeout/flow_graceful_timeout.py
Timeout handler 成功时，Dex 会应用暂存的 write、best-effort Channel deletion、Channel publication 和 decision。如果它耗尽配置的重试，Dex 会让 Flow 失败。