--- title: "Step" description: "Step 是 Flow 里的一个后台工作单元。它可以先 durable 地等待某件事发生，再执行工作，然后决定下一步运行什么。Dex 在你的 Worker 上运行 Step，并记录每个结果。" canonical_url: "https://docs.superdurable.io/zh-Hans/concepts/step/" source: "docs/i18n/zh-Hans/docusaurus-plugin-content-docs/current/concepts/step/index.mdx" locale: "zh-Hans" --- # Step **Step** 是 [Flow](https://docs.superdurable.io/zh-Hans/concepts/flow.md) 里的一个后台工作单元。它可以先 durable 地等待某件事发生，再执行工作，然后决定下一步运行什么。Dex 在你的 Worker 上运行 Step，并记录每个结果。 Step 由一个原生 struct 或 class 实现，它的类型就是 **StepType**。同一个 Step type 在一个 Flow 里可以运行多次，甚至并行运行。每次运行是一个 **StepExecution**，用 **StepExecutionID** 标识，比如 **Charge-2**。 ## Step input 每个 Step execution 有自己的 typed input，Dex 把它和 execution 一起保存。input 用来区分同一个 Step type 的不同 execution。start Step 的 input 就是 Flow 的 start input。 ## WaitFor 和 Execute 一个 Step 有一个或两个方法。 - **WaitFor** 可选。它返回一个 durable 的 **Wait**。Dex 保存这个 wait，替你检查条件，条件满足后调用 **Execute**。 - **Execute** 必需。它执行工作，并返回一个 **StepDecision**。 Step 没有 **WaitFor** 时，**Execute** 立即运行。 ExampleStep 的 WaitFor 返回一个 Wait，Dex 评估其中的 Timer、Channel 或 SubFlow 条件。条件满足后，Execute 返回 GoTo(Complete Flow, input + 1)。StepDecision 可以是 GoTo、GracefulComplete 或 ForceFail。 下面这个 Step 等待一条审批消息，然后进入第二个 Step，由它完成 Flow。 **Python** ```python approval = Channel("Approval", str) class StepSecond(Step[int]): def execute(self, context: Context, input: int) -> StepDecision: return graceful_complete(input + 1) class Example(Step[int]): def __init__(self, second: StepSecond) -> None: self.second = second def wait_for(self, context: Context, input: int) -> Wait: return Wait.until(approval.for_one()) def execute(self, context: Context, input: int) -> StepDecision: return go_to(StepSecond, input + 1) ``` 例子: [examples/python/dex_examples/primitives/step/step_flow.py](https://github.com/superdurable/dex/tree/main/examples/python/dex_examples/primitives/step/step_flow.py) **Go** ```go var Approval = dex.DefineChannel[string]("Approval") type Example struct { dex.StepDefaults } func (Example) WaitFor(_ dex.Context, _ int) (*dex.Wait, error) { return dex.Until(Approval.ForOne()), nil } func (Example) Execute(_ dex.Context, input int) (*dex.StepDecision, error) { return dex.GoTo(stepSecond{}, input+1), nil } type stepSecond struct { dex.StepDefaultsNoWaitFor[int] } func (stepSecond) Execute(_ dex.Context, input int) (*dex.StepDecision, error) { return dex.GracefulComplete(input + 1), nil } ``` 例子: [examples/go/primitives/step/workflow.go](https://github.com/superdurable/dex/tree/main/examples/go/primitives/step/workflow.go) **Java** ```java final class Example implements Step { @Override public Class getInputType() { return Integer.class; } @Override public Wait waitFor(final Context context, final Integer input) { return Wait.until(approval.forOne()); } @Override public StepDecision execute(final Context context, final Integer input) { return StepDecision.goTo(StepSecond.class, input + 1); } } static final class StepSecond implements Step { @Override public Class getInputType() { return Integer.class; } @Override public StepDecision execute(final Context context, final Integer input) { return StepDecision.gracefulComplete(input + 1); } } ``` 例子: [examples/java/src/main/java/io/superdurable/dex/primitives/step/StepFlow.java](https://github.com/superdurable/dex/tree/main/examples/java/src/main/java/io/superdurable/dex/primitives/step/StepFlow.java) **TypeScript** ```typescript const approval = new Channel("Approval", stringCodec); class Example implements Step { public readonly inputCodec = doubleCodec; public constructor(private readonly flow: StepFlow) {} public getStepType(): string { return "Example"; } public waitFor(_context: Context, _input: number): Wait { return Wait.until(approval.forOne()); } public execute(_context: Context, input: number): StepDecision { return goTo(StepSecond, input + 1); } } class StepSecond implements Step { public readonly inputCodec = doubleCodec; public getStepType(): string { return "StepSecond"; } public execute(_context: Context, input: number): StepDecision { return gracefulComplete(input + 1); } } ``` 例子: [examples/typescript/src/primitives/step/step-flow.ts](https://github.com/superdurable/dex/tree/main/examples/typescript/src/primitives/step/step-flow.ts) **Rust** ```rust static APPROVAL: LazyLock> = LazyLock::new(|| Channel::new("Approval")); #[derive(Default)] struct Example; impl Step for Example { type Input = i32; fn wait_for(&self, _context: &mut Context, _input: Self::Input) -> HandlerResult { Ok(Wait::until(APPROVAL.for_one())) } fn execute(&self, _context: &mut Context, input: Self::Input) -> HandlerResult { Ok(StepDecision::go_to(&StepSecond, input + 1)) } } #[derive(Default)] struct StepSecond; impl Step for StepSecond { type Input = i32; fn execute(&self, _context: &mut Context, input: Self::Input) -> HandlerResult { Ok(StepDecision::graceful_complete(input + 1)) } } ``` 例子: [examples/rust/src/primitives/step/flow.rs](https://github.com/superdurable/dex/tree/main/examples/rust/src/primitives/step/flow.rs) ## 等待 等待不占用 Worker 或线程。Dex 把它保存在 Flow 的状态里，条件满足时唤醒 Step。Step 可以等待三类条件： - [Timer](https://docs.superdurable.io/zh-Hans/concepts/primitives/timer.md) 触发； - [Channel](https://docs.superdurable.io/zh-Hans/concepts/primitives/channel.md) 上有消息到达； - [SubFlow](https://docs.superdurable.io/zh-Hans/concepts/primitives/subflow.md) 完成。 用 **AnyOf** 或 **AllOf** 组合条件。**AnyCombinationOf** 在几组条件中任意一组全部满足时成立。等待结束后，**Execute** 可以读取满足它的结果，比如它消费的 Channel 消息。见 [Waits](https://docs.superdurable.io/zh-Hans/reference/python/waits.md)。 要等待并行分支，用 **Join**。它不是另一类条件：它等待的是各分支上报结果的 ChannelMap。见 [等待并行 Step](https://docs.superdurable.io/zh-Hans/design-patterns/parallel/await.md)。 ## StepDecision **Execute** 返回一个 **StepDecision**。它可以进入下一个 Step，也可以同时进入多个 Step。分支、循环和并行工作都这样构建。 GoTo：ExampleStep 返回 GoTo(Complete Flow)，完成 Flow。GoToMany：ExampleStep 启动 StepA 和 StepB，两者并行运行。循环：ExampleStep 转到 ProcessStep，ProcessStep 再回到 ExampleStep。 决定也可以关闭 Flow 或结束一个分支： - **GracefulComplete** 请求 Dex 在没有其他 active Step 时完成 Flow。 - **ForceComplete** 立即完成 Flow。 - **ForceFail** 立即让 Flow 失败。 - **DeadEnd** 结束这个分支，但不关闭 Flow。 完成类的决定可以返回 Flow output，client 把它作为 Flow result 读取。多个完成类的决定都返回 output 时，result 会保留每一个。只有最终完成的 Flow 才保留这些 output。如果 Flow 没有完成（失败、被取消、被终止或超时），它的 result 不含任何 output。 决定还可以取消其他 active Step。见 [Parallel Steps](https://docs.superdurable.io/zh-Hans/design-patterns/parallel.md) 和 [Parallel SubFlows](https://docs.superdurable.io/zh-Hans/design-patterns/parallel-subflows.md) 模式。 ## StepOptions Step 返回 **StepOptions**，控制它的方法如何运行。**WaitFor** 和 **Execute** 各有自己的设置。 - **Timeout** 限制方法的一次 attempt。 - **Retry policy** 在 attempt 失败后再次运行同一个方法。 - **Failure policy** 决定重试用完之后怎么办。默认让 Flow 失败。**Execute** 可以改为进入一个 recovery Step，**WaitFor** 可以继续执行 **Execute**。见 [WaitFor failure policy](https://docs.superdurable.io/zh-Hans/reference/python/failure-and-retries.md#waitfor-failure-policy)、[Execute failure policy](https://docs.superdurable.io/zh-Hans/reference/python/failure-and-retries.md#execute-failure-policy) 和 [Last failure info](https://docs.superdurable.io/zh-Hans/reference/python/failure-and-retries.md#last-failure-info)。 - **Durability** 决定 Dex 什么时候存储结果。默认的 **SYNC** 在下一步工作开始前存储。**ASYNC** 更快，但 server 故障可能让方法再运行一次。 - **Heartbeat** 让长时间运行的方法报告进度并保存 checkpoint。 - **State loading** 和 **Attribute locks** 决定方法加载哪些状态、锁住哪些 Attribute。 **StepDecision** 可以为某一次 Step execution 覆盖这些选项。 ## 保证 Dex 单独保存每次 **WaitFor** 或 **Execute** 调用。它把这次调用的 Attribute 写入和 Channel 发布与结果一起保存，而且只在方法成功时保存。对外部系统的调用不在这次保存里。 一个 Step 方法的修改会一次全部生效。但这挡不住其他 Step 和 RPC 同时修改同样的状态。方法读到的是 Dex 把它交给 Worker 时的 Flow 状态。在方法的修改生效之前，其他 Step 和 RPC 可能已经改了这些状态。Dex 不检查方法读到的值是否仍然是最新的。所以两个先读后写同一个 Attribute 的 Step 或 RPC，可能互相覆盖对方的更新。要避免这种情况，让它们都在这个 Attribute 上声明 lock。Dex 会让它们依次执行。见 [Attribute lock](https://docs.superdurable.io/zh-Hans/concepts/primitives/attribute.md#locking)。 一个 Step 方法可能运行不止一次：失败后的重试会再运行它。Dex 不一定知道失败的 attempt 有没有执行工作。请让每个外部副作用都幂等，用 Flow ID 加 StepExecutionID，或你自己的业务 key 作为幂等 key。Dex 无法判断的情况见 [Failure and redelivery](https://docs.superdurable.io/zh-Hans/reference/python/failure-and-retries.md#failure-and-redelivery)。 重试用完时，recovery Step 能知道工作是肯定没有运行，还是可能运行过。 取消一个 Step，会丢弃它当前方法还没有保存的一切：Attribute 写入、Channel 发布和它的决定。它的 **WaitFor** 已经保存的内容保留。一个决定或 RPC 永远不会取消它自己启动的 Step。被取消的 Step 已经消费的消息会怎样，见 [Channel 的保证](https://docs.superdurable.io/zh-Hans/concepts/primitives/channel.md#%E4%BF%9D%E8%AF%81)。 Dex 不会强行停止被取消方法的代码。方法应检查取消状态并返回。在它返回之前，它的代码会继续运行。它对外部系统的调用仍然会生效，即使 Flow 已经关闭。 **GracefulComplete** 结束它自己的分支，并请求 Dex 完成 Flow。一旦没有 active Step，Dex 就开始完成 Flow。在此之前，其他 Step 和 RPC 仍然可以启动新的 Step，这些 Step 也会推迟完成。之后某个分支以 **DeadEnd** 结束，也不会撤回这个请求。最后一个 Step 结束后，Dex 还要处理完待写的 Attribute Sync 和已经开始的 transactional RPC，所以 Flow 可能会短暂保持 **RUNNING**。 没有待处理的 **GracefulComplete** 时，一个没有 active Step 的 Flow 会继续运行。RPC 仍然可以在里面启动新的 Step。用一个完成类的决定结束最后一个分支。如果 Flow 不能无限等待，就给它设置 Flow timeout。 Dex 大致按 Step 就绪的顺序启动它们，但不严格保证。不要依赖并行 Step 启动或结束的顺序。顺序重要时，用 Channel 或 Attribute lock。 ## 参考 英文的 SDK 参考包含： - [Step 方法](https://docs.superdurable.io/zh-Hans/reference/python/step.md)，以及每个方法提交什么； - [等待](https://docs.superdurable.io/zh-Hans/reference/python/waits.md)、条件结果和 [AnyCombinationOf](https://docs.superdurable.io/zh-Hans/reference/python/waits.md#anycombinationof)； - [Step decision](https://docs.superdurable.io/zh-Hans/reference/python/step-decisions.md)，包括 [ForceCompleteIfChannelsEmpty](https://docs.superdurable.io/zh-Hans/reference/python/step-decisions.md#forcecompleteifchannelsempty)； - [重试和失败策略](https://docs.superdurable.io/zh-Hans/reference/python/failure-and-retries.md#retry-and-failure-policies)、[durability](https://docs.superdurable.io/zh-Hans/reference/python/step-options.md#durability)、[超时](https://docs.superdurable.io/zh-Hans/reference/python/step-options.md#timeouts-and-total-duration)、[heartbeat](https://docs.superdurable.io/zh-Hans/reference/python/step-options.md#heartbeat) 和[取消](https://docs.superdurable.io/zh-Hans/reference/python/step-options.md#cancellation)； - [状态加载](https://docs.superdurable.io/zh-Hans/reference/python/step-options.md#selective-state-loading)和 [Attribute locks](https://docs.superdurable.io/zh-Hans/reference/server/runtime-data.md#attribute-locks)； - [失败与重新投递](https://docs.superdurable.io/zh-Hans/reference/python/failure-and-retries.md#failure-and-redelivery)。