Continue-As-New 是什么？

Continue-As-New 支持成功关闭工作流执行，同时创建新工作流执行。可以将其视为工作流执行时间过长或接近某些扩展限制时的检查点。

新工作流执行位于同一链中；保留相同的工作流 ID，但获得新运行 ID 和全新事件历史。同样接收工作流的参数。

如何使用 Python SDK Continue-As-New？

首先，合理设计工作流参数，以便通过 Continue-As-New 进入下个工作流运行时，能传入“当前状态”。对于工作流的原始调用者而言，该状态通常设置为 None。

@dataclass
class ClusterManagerInput:
    state: Optional[ClusterManagerState] = None
    test_continue_as_new: bool = False

@workflow.run
async def run(self, input: ClusterManagerInput) -> ClusterManagerResult:

上述代码片段中的测试钩子将在下文介绍。

在工作流内部，使用相同的类型调用 continue_as_new() 函数。这将立即停止当前工作流，启动新工作流。

workflow.continue_as_new(
  ClusterManagerInput(
      state=self.state,
      test_continue_as_new=input.test_continue_as_new,
  )
)

包含消息处理器的工作流注意事项

如果使用更新或信号，那么不要在处理器中调用 Continue-As-New。相反，应在主工作流中等待处理器完成后，再执行 continue_as_new。请参阅 all_handlers_finished 示例。

何时适合使用 Python SDK 执行 Continue-As-New？

当工作流可能遇到性能下降或事件历史限制时，应使用 Continue-As-New。

Temporal 根据这些限制追踪工作流的进度，以便让开发者知道何时应执行 Continue-As-New。调用 workflow.info().is_continue_as_new_suggested() 检查是否已到该时机。

如何使用 Python SDK 测试 Continue-As-New？

对执行 Continue-As-New 的工作流进行测试，可能比较耗时且消耗大量资源。因此，可以添加测试钩子，在自动化测试中更快地验证工作流的 Continue-As-New 行为。

比如，当 test_continue_as_new == True 时，此示例将创建仅用于测试的变量 self.max_history_length，并且将其设置为较小的值。工作流中的辅助方法在每次考虑使用 Continue-As-New 时检查该变量：

def should_continue_as_new(self) -> bool:
    if workflow.info().is_continue_as_new_suggested():
        return True
    # For testing
    if (
        self.max_history_length
        and workflow.info().get_current_history_length() > self.max_history_length
    ):
        return True
    return False