跳到主要内容

Flow options

Dex 允许你自定义 Flow execution 的行为。

  • StartFlowOptions 在客户端启动 Flow（或 SubFlow）时生效，启动后不可变。
    • 它们控制 timeout、ID reuse、初始 Attribute、request 幂等，以及该次 execution 的可选 config override。
  • FlowConfig 提供运行时行为，启动后可变。
    • 包括 Step durability 默认值、Worker 路由、active-Step 搜索索引与 Attribute Store projection target。

StartFlowOptions​

字段作用
TimeoutFlow timeout。省略或设为零可禁用。
TimeoutPolicyFlow timeout 触发时的行为。
TimeoutHandlerOptionsTimeout handler 的 execution、retry、recovery、durability、locking 与 state loading。
IDReusePolicy启动时如何复用已有 Flow ID。
RequestID重试同一次启动请求的幂等键。
AlreadyStartedOptions如何处理已启动的 Flow。
RetryPolicy不支持。server-zd 会以 INVALID_ARGUMENT 拒绝它。见 RetryPolicy。
StartDelay不支持。server-zd 会以 INVALID_ARGUMENT 拒绝正值。见 StartDelay。
Attributes初始 Attribute 值。
FlowConfig每次 execution 的 Flow 配置。

如需在 Flow 截止时间执行应用定义的处理，请参阅 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")

例子: examples/python/dex_examples/primitives/flow/controller.py

Timeout 与 TimeoutPolicy​

Timeout 是整个 Flow 的一个 durable timer。只有 Timeout 为正数时才能设置 TimeoutPolicy。

  • FAIL — 以 timeout error 失败 Flow。
  • CANCEL — 取消 Flow，不会重试。
  • HANDLER — 运行一次逻辑 Flow timeout-handler execution。

提供正数 Timeout 时，Dex SDK 默认使用 FAIL。Flow 有 timeout handler 时，则使用 HANDLER。没有 handler 时使用 HANDLER 无效。

只有 Timeout 为正数且最终 policy 为 HANDLER 时，才能设置 TimeoutHandlerOptions。它采用 Execute 语义：method timeout 限制单次 attempt，Retry 可以运行多次 attempt，HeartbeatTimeout 检测停滞的 attempt，Failure 可以在 retry 耗尽后进入已注册的无输入 Step。Recovery Step 会收到 null 或 unit input，并从 handler Context 读取最终 error。

Handler 的 attempt 与 retry 计时从 soft Flow timeout 触发时开始，因此可以超过原始 Flow deadline。未配置 failure target 时，handler retry 耗尽会使 Flow 失败。

普通 Attribute 和所有 Channel size metadata 会自动加载。AttributeMap value 与 pending Channel message 必须通过 TimeoutHandlerOptions 显式选择。Handler 可以先暂存 Attribute write、Channel deletion 与 Channel publication，再返回 decision。Dex 没有 Flow 级别的 retry，因此每个 Flow run 只执行一次 timeout handler。

把 TimeoutPolicy 设为 HANDLER，timeout 触发时就会运行下面的代码。这个 handler 记录 Flow 结束的原因，再用应用自己的原因失败该 Flow。

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

例子: examples/python/dex_examples/primitives/flow/example_flow.py

Flow ID、RequestID 与 AlreadyStartedOptions​

IDReusePolicy 决定这个 Flow ID 已存在时该怎么做。RequestID 表示这次调用是否在重试同一次启动请求。

这些选项如何一起工作​

这些选项依次回答不同的问题：

  1. RequestID 标识一次 start request。省略时，SDK 会生成新的 UUID，所以另一次 StartFlow 调用是新的 request。只在重试完全相同的 start 时复用它。Dex 比较的是 RequestID，不比较 start input 或 options。
  2. IDReusePolicy 决定这个 request 能否用该 Flow ID 启动另一个 run。
  3. AlreadyStartedOptions.IgnoreError 只在 ID reuse policy 因 already-started 拒绝启动时生效。只有 RequestID 相同时，它才返回已有的 Run ID。
重试请求AlreadyStartedOptions.IgnoreErrorID reuse policy 拒绝启动时
未提供 RequestID关闭或启用SDK 发送新 UUID，Dex 返回 already-started error。
不同的 RequestID关闭或启用Dex 返回 already-started error。
相同 RequestID关闭Dex 返回 already-started error。
相同 RequestID开启Dex 返回已有 Run ID。

ID reuse policy 决定 Dex 是否会首先拒绝启动：

IDReusePolicy已有 active Flow已有 closed Flow
IDReuseDefault拒绝启动。RequestID 相同且启用 IgnoreError 时返回 active Run ID。拒绝启动。RequestID 相同且启用 IgnoreError 时返回已有 Run ID。
IDReuseAllowIfPreviousFailed拒绝启动。RequestID 相同且启用 IgnoreError 时返回 active Run ID。只有上一次 run 失败后才启动新的 run；否则拒绝。
IDReuseAllowIfNotRunning拒绝启动。RequestID 相同且启用 IgnoreError 时返回 active Run ID。启动新的 run。
IDReuseDisallow拒绝启动。RequestID 相同且启用 IgnoreError 时返回已有 Run ID。拒绝启动。RequestID 相同且启用 IgnoreError 时返回已有 Run ID。
IDReuseTerminateIfRunning终止 active run 并启动 replacement。不会产生 already-started error。启动新的 run。

Dex Server 的逻辑​

Dex Server 要求 RequestID 非空。省略时，SDK 会在发送请求前生成它。服务端将它记录在 Flow execution 中，再使用请求的 ID reuse policy 让已配置的 backend 启动 Flow。IDReuseDefault 使用服务端默认行为：拒绝所有曾经使用过的 Flow ID。需要让已关闭的 Flow ID 启动新 execution 时，必须显式选择 IDReuseAllowIfNotRunning。

如果 backend 返回 Flow 已启动，Dex 会检查 AlreadyStartedOptions.IgnoreError。启用时，Dex 读取已有 run 记录的 RequestID。只有两个 RequestID 相同时才返回该 Run ID；否则仍然返回 already-started error。

StartFlow 最佳实践​

使用稳定的业务 Flow ID 直接调用 StartFlow。不要仅为了避免可能的重试冲突，在启动前读取 snapshot、搜索 Flow 或查询状态。启动成功后，也不必回读以重复核对提交的身份、owner 或初始 Attributes。

同一个逻辑启动请求的重试，应同时使用稳定的 RequestID、IDReuseDisallow 和忽略 already-started 的选项。每次重试保持相同输入和 options。IDReuseDisallow 防止 Flow 关闭后再次启动；允许新 run 的策略可能绕过匹配 RequestID 的处理路径。

下面的 helper 接收已验证的整数输入和稳定的 Flow ID。这就是此例的完整业务输入，两者共同确定 RequestID，其余启动值固定不变。实际应用应包含所有会改变操作的逻辑输入，例如 tenant、owner、mode、revision 或初始业务 Attributes。Dex 比较的是 RequestID，不会比较 payload 是否相等。逻辑请求改变时，不要复用同一个 RequestID。

每个 helper 在启动被接受后返回已知的业务 Flow ID，并保留错误供调用方处理。启动前后都没有读取。返回的 ID 表示请求已被接受，不是当前业务 snapshot，也不证明 Flow 已完成。

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

例子: examples/python/dex_examples/primitives/flow/controller.py

例如，用输入 7 启动 invoice-42 时，RequestID 是 start-example-flow:7:invoice-42。只要 backend 仍保留该 execution，重试相同调用就会返回它，关闭后也一样。用输入 8 启动相同 Flow ID 会生成不同的 RequestID，并返回 already-started error。成功路径无需再比较 snapshot。

响应承诺业务里程碑时才等待​

API 如果必须确认一个重要操作已经完成，应在 StartFlow 成功后直接调用 WaitForAttributeMatch 或 WaitForStepCompletion，匹配 RequestID 的去重成功也走同一条等待路径。去重只确认同一个启动已被接受，不证明操作已完成。启动和等待之间不需要读取以重复验证身份。

例如，API 承诺返回时，业务记录已经提交到作为 source of truth 的数据库表。Owning Flow 在 Execute 中写入该记录，然后记录已提交的 Attribute，或完成持久化 Step。API 等待该 Attribute 的已提交值或指定的 StepExecutionID，再返回已确认的持久化结果。标记必须在数据库提交成功后才发布。WaitForStepCompletion 只确认该 Step 已完成，不返回 Step 输出，也不等待整个 Flow 完成。

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

只等待能够证明响应所承诺结果的条件。如果响应只确认后台工作已被接受，启动后就直接返回，不添加等待。使用调用方 deadline 或 request timeout 限制必要的等待。等待超时不会撤销启动，也不证明数据库写入失败；应按 API 契约返回完成状态仍为 pending 或未知，并在重试时保留同一个启动身份。

各语言 API 见 Step 完成等待 和 Attribute 匹配等待。Attribute 等待针对 active run。如果 Flow 在等待建立前已关闭，应根据实际等待错误，从权威业务记录或支持的终态读取确认结果，不要假设 Attribute 等待可以查询已关闭的 run。

只有错误结果需要时才读取​

错误处理由业务契约决定。如果业务接受将 already-started error 直接转换为 ALREADY_EXISTS，就直接返回冲突。不要仅因为出现错误就增加 read-only RPC。上面的 helper 保留 typed error，交由应用边界转换。

只有读取会改变响应或下一步业务决策时才读取。例如，API 明确允许接续已有操作时，可以在冲突后调用它的 typed read-only RPC，判断已有操作是否满足请求。API 如果承诺返回当前 snapshot，就必须读取当前状态，不能把提交的初始 Attributes 当成当前 snapshot。启动被接受不代表业务已完成。

情况处理方式
启动被接受，响应只确认接受请求返回 Flow ID 和约定的接受字段，不读取 snapshot。
启动成功或去重成功，响应承诺关键操作已完成等待已提交的 Attribute 条件或指定 Step 完成，再返回该里程碑结果。
已启动，业务接受 ALREADY_EXISTS直接返回冲突，不读取 snapshot。
已启动，业务允许接续已有操作仅为判断契约是否允许接续，读取 typed snapshot。
响应丢失，是否已接受启动尚不确定使用相同身份和 options 有界重试；或者在权威状态能够确认结果时读取。

不要无限重试已确认的冲突。如果没有传入稳定的 RequestID，单独调用 SDK 时会生成新 ID；仅启用忽略选项不能为这些调用去重。

依赖启动由 owning Flow 持久化协调​

一个业务操作由 API 启动一个 owning Flow，后续工作交给它的持久化 Steps。不要在 HTTP handler 中启动第一个 Flow、读取其状态、再启动依赖的 Flow。如果进程在两次启动之间退出，就没有持久化 Step 负责补上第二次启动。

共享一个身份和生命周期的工作应使用 Steps。确实需要独立 owner 和生命周期的 top-level Flow 时，在 Execute Step 中使用稳定的下游 Flow ID、RequestID 和启动 options 创建它，并将启动输入保存在 owning Flow 中。如果 Worker 在下游启动被接受后、Step 提交前退出，Step 可以重试同一个启动请求而不创建重复 execution。WaitFor 用于持久化条件。一个独立 owner 的 top-level Flow 不会仅因另一个 Flow 启动它，就需要改成 SubFlow。

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 不支持 Flow 级别的 RetryPolicy。设置了它的启动请求会以 INVALID_ARGUMENT 失败。由 issue 996 跟踪。需要重试工作时，请使用受支持的 Step retry。

StartDelay​

server-zd 不支持 StartDelay。带正值延迟的启动请求会以 INVALID_ARGUMENT 失败。由 issue 995 跟踪。需要延迟第一项工作时，请不带延迟启动 Flow，并让 StartStep 以 durable Timer 等待开头。

Attributes​

Attributes 写入初始值。每个值必须匹配 Flow persistence schema 中的 Attribute 或 Attribute map。

FlowConfig​

在 StartFlowOptions 中使用 ConfigOverride 修改一个 Flow 的初始配置。它只修改提供的字段，不会整体替换 FlowConfig。

启动时，Dex 会把提供的字段覆盖到 Dex Server 默认值上。省略的字段保留默认值。

字段行为
ActiveStepSearchMode索引所有 active Step、只索引运行过 WaitFor 的 Step，或不索引 active Step。内置 Server 默认不索引 active Step；如需启用，请显式设置模式。
ContinueAsNewThreshold已弃用。Dex 不会把 Flow continue as new。Dex 会拒绝负值，把值保存在 FlowConfig 中，其他情况下忽略它。ContinueAsNewPageSizeBytes 的行为相同。
StepDurability没有 Step override 的 method 使用的默认 durability。Dex 依次使用 method override、这里的值，最后使用 SYNC。
AttributeStoreNamesDex Server Attribute Store 名称，供 opt-in attribute sync 的 Attribute 使用。

用 Client 更新 FlowConfig​

Flow active 时可以调用 UpdateFlowConfig。只发送要修改的字段。它们会与已有配置合并。SDK 的 UpdateFlowConfig 调用不会改变 Flow 的 Worker 路由。