4.9 - 版本管理

由于 Temporal 中的工作流执行可能持续很长时间——有时长达数月甚至数年——因此，常常需要对工作流定义进行修改，包括正在进行的工作流执行。

Temporal 平台要求工作流代码具备确定性。更改工作流代码可能在重放时导致非确定性行为，此时需要选择一种版本管理方法，优雅地更新正在运行的工作流。此要求仅适用于工作流编排逻辑。非确定性工作，比如 API 调用、数据库查询，应放在 Activity 中执行，由 Temporal 可靠地重试。

借助版本管理，可以修改工作流定义，使新工作流执行使用更新后的代码，现有的工作流执行继续运行原始版本。版本管理方法包括：

Worker 版本管理。Worker 版本管理功能支持为 Worker 添加标签，以编程方式在版本化部署中逐步推出更新，从而让旧 Worker 继续运行旧代码路径，而新 Worker 运行新代码路径。

通过补丁进行版本管理。该方法通过在代码中添加与特定修订版本对应的分支来实现。它能够将代码变更应用于新工作流执行，同时避免对正在进行的工作流执行造成破坏性影响。

Worker 版本管理

Worker 版本管理功能支持为 Worker 添加标签，以编程方式在版本化部署中逐步推出更新，从而让旧 Worker 继续运行旧代码路径，而新 Worker 运行新代码路径。使用这种方式，可以将工作流锁定到特定修订版本，从而无需补丁操作。

通过补丁进行版本管理

添加补丁

补丁为特定变更在工作流中定义逻辑分支，类似于功能开关（Feature Flag）。它能够将代码变更应用于新工作流执行，同时避免对正在运行的工作流执行造成破坏性影响。当需要对代码进行实质性修改，且这些修改可能影响现有工作流执行时，那么可以创建补丁。

假设初始工作流版本名为 pre_patch_activity：

from datetime import timedelta
from temporalio import workflow

with workflow.unsafe.imports_passed_through():
from activitiesimport pre_patch_activity

@workflow.defn
class MyWorkflow:
    @workflow.run
    async def run(self) -> None:
        self._result =await workflow.execute_activity(
            pre_patch_activity,
            schedule_to_close_timeout=timedelta(minutes=5),
        )

    @workflow.query
    def result(self) -> str:
        return self._result

现在，希望更新代码，改为运行 post_patch_activity。该版本是期望的最终状态。

from datetime import timedelta
from temporalio import workflow

with workflow.unsafe.imports_passed_through():
from activitiesimport post_patch_activity

@workflow.defn
class MyWorkflow:
    @workflow.run
    async def run(self) -> None:
        self._result =await workflow.execute_activity(
            post_patch_activity,
            schedule_to_close_timeout=timedelta(minutes=5),
        )

    @workflow.query
    def result(self) -> str:
        return self._result

问题在于，在确认没有更多使用 pre_patch_activity 代码创建的工作流仍在运行之前，不能直接部署 post_patch_activity，否则很可能导致非确定性错误。相反，需要部署 post_patched_activity，并且使用 patched 函数决定应该执行哪个版本的代码。

补丁操作分为三个步骤：

在更新代码中，使用 patched() 函数打补丁，使新旧代码并行运行。

移除旧代码，使用 deprecate_patch() 将特定补丁标记为已废弃。

当不再有任何使用之前版本代码的工作流执行处于开启状态时，移除 deprecate_patch() 调用。

下面按顺序逐步讲解该过程。

在新代码中打补丁

使用 patched 会在事件历史中插入标记。在重放过程中，如果 Worker 遇到带有该标记的历史，而当前工作流代码不生成相同的补丁标记（本例中为 my-patch），那么工作流任务将失败。这确保可以将 post_patch_activity 的代码作为“功能开关”安全地与原始版本（pre_patch_activity）并行部署。

from datetime import timedelta
from temporalio import workflow

with workflow.unsafe.imports_passed_through():
    from activities import post_patch_activity, pre_patch_activity


@workflow.defn
class MyWorkflow:
    @workflow.run
    async def run(self) -> None:
if workflow.patched("my-patch"):
            self._result =await workflow.execute_activity(
                post_patch_activity,
                schedule_to_close_timeout=timedelta(minutes=5),
            )
else:
            self._result =await workflow.execute_activity(
                pre_patch_activity,
                schedule_to_close_timeout=timedelta(minutes=5),
            )

    @workflow.query
    def result(self) -> str:
        return self._result

废弃补丁

当确认所有使用 pre_patch_activity 代码启动的工作流均已超出保留期后，即可废弃该补丁。

当工作流不再运行打补丁前的代码路径时，便可以部署包含 deprecate_patch() 的代码。这些 Worker 将运行最新版本的工作流代码，不再需要该补丁。已废弃的补丁充当补丁流程的最后阶段与最终无补丁状态之间的桥梁。它们与常规补丁类似，也在事件历史中添加标记。但不同的是，当工作流代码不生成该标记时，不会导致重放失败。

from datetime import timedelta
from temporalio import workflow

with workflow.unsafe.imports_passed_through():
    from activities import post_patch_activity


@workflow.defn
class MyWorkflow:
    @workflow.run
    async def run(self) -> None:
        workflow.deprecate_patch("my-patch")
        self._result = await workflow.execute_activity(
            post_patch_activity,
            schedule_to_close_timeout=timedelta(minutes=5),
        )

    @workflow.query
    def result(self) -> str:
        return self._result

移除补丁

当所有在补丁生命周期内启动的工作流均已超出保留期后，即可安全地部署不再使用 patched() 或 deprecate_patch() 调用的 Worker：

# ...
@workflow.defn
class MyWorkflow:
    @workflow.run
    async def run(self) -> None:
        self._result = await workflow.execute_activity(
            post_patch_activity,
            schedule_to_close_timeout=timedelta(minutes=5),
        )

补丁支持对正在运行的工作流进行更改。可以在引入兼容性变更的同时，避免引入非确定性错误。

工作流切换

演示如何对整个工作流进行切换对理解补丁为何有用很有帮助。

由于不兼容的变更只影响同一类型的正在执行的工作流，因此可以通过在变更时，创建全新工作流的方式，避免确定性错误。具体做法是：复制工作流定义函数，为其赋予不同的名称，将两个名称都注册到 Worker 中。

比如将 PizzaWorkflow 复制为 PizzaWorkflow2：

@workflow.defn(name="PizzaWorkflow")
class PizzaWorkflow:
    @workflow.run
    async def run(self, name: str) -> str:
        # this function contains the original code

@workflow.defn(name="PizzaWorkflowV2")
class PizzaWorkflowV2:
    @workflow.run
    async def run(self, name: str) -> str:
        # this function contains the updated code

然后，需要更新 Worker 配置以及任何其他标识符字符串，同时注册两种工作流类型：

worker = Worker(
    client,
    task_queue="your-task-queue",
    workflows=[PizzaWorkflow, PizzaWorkflowV2],
)

这种方法的缺点在于需要复制代码，以及更新用于启动工作流的所有命令。随着时间的推移，这种方式变得不切实际。此外，这种方法也无法对仍在运行的工作流进行版本管理——它本质上只是切换，与补丁（Patching）不同。