工作流协议
工作流构建块底层机制的详细描述。
本文档从底层详细说明 Dapr 工作流协议与运行时约定。面向的读者是构建工作流 Worker 的 SDK 作者以及演进 Dapr 边车工作流引擎的运行时维护者。
概述
Dapr 工作流采用"边车即调度器"模式：Dapr 运行时（边车）作为工作流引擎，应用 SDK 作为工作流 Worker。所有控制与执行流量均通过 gRPC 传输。
协议界面分为两类：
- 管理 API（通过 SDK 可访问的标准 Dapr gRPC）：
- 启动、终止、暂停、恢复、重新运行、清除以及查询工作流实例。
- 执行 API（Task Hub 协议）：
- Worker 侧接口，用于接收编排/活动工作项并上报完成（例如通过
TaskHubSidecarService）。
核心组件
工作流引擎（Dapr 边车）
管理工作流状态转换、历史持久化、编排与活动任务的调度，以及可靠交付语义。默认情况下，利用 Dapr Actors 作为后端，实现可持久化、分区的执行。
工作流 Worker（应用 SDK）
连接到边车，轮询编排与活动工作项，执行用户定义的逻辑，并将结果、失败与心跳返回给引擎。编排逻辑必须是确定性的；活动逻辑无需是确定性的。
编排
定义工作流的确定性协调器。引擎通过重放历史来驱动编排，重建状态并调度出站任务（活动、子编排、定时器、外部事件）。
活动
原子工作单元。活动执行保证至少一次，并将结果或失败报告回引擎。建议幂等，并在上下文中提供任务执行标识符以辅助实现幂等。
状态存储与后端
工作流历史与状态持久化保存。引擎通常在所选持久化层上实现任务中心模式，并使用 Dapr Actors 作为默认的可靠性底层。
执行模型
Dapr 工作流基于持久任务框架（DTFx）执行语义：
基于重放的执行
编排器从其事件历史重放，以重建确定性状态。所有非确定性操作（时间、随机值、I/O）必须由引擎中介（例如通过定时器、活动调用、外部事件）。
确定性编排器
除引擎中介效果外，编排器代码必须无副作用。控制流在重放期间必须可复现。
活动至少一次交付、状态提交恰好一次
活动可能被多次传递。引擎确保工作流状态提交是幂等的，并恰好一次应用。
边车即调度器
边车拥有调度权，并在向 Worker 分发工作前持久化所有历史/事件。从引擎视角看，Worker 是无状态执行器。
协议界面
- 管理 API（标准 Dapr gRPC）
- 启动工作流：创建并持久化初始历史事件；返回实例元数据。
- 终止 / 暂停 / 恢复：通过持久化的控制事件驱动生命周期转换。
- 查询：检索实例状态、历史、输出、失败详情及自定义元数据。
- 重新运行：从历史事件启动新的工作流实例。
- 清除：主动清除工作流历史与状态。
注意：关于确切的 RPC 形状、错误码和语义，参见 管理 API 规范。
- 执行 API（Task Hub 协议）
- 轮询工作：Worker 获取编排与活动工作项。
- 完成 / 失败工作：Worker 报告完成结果或失败；引擎将这些追加到历史并推进编排进度。
- 心跳 / 租约：用于长时间运行的活动与协作式重新平衡的可选机制。
- 定时器与外部事件：作为历史事件传递给编排，以保持重放的确定性。
注意：参见定义了 TaskHubSidecarService 约定、负载架构与排序规则的 执行 API 规范。
请求与运行时生命周期
- 启动工作流
- 客户端通过管理 API 调用
StartWorkflow。 - 引擎持久化初始事件（例如
ExecutionStarted）并实例化实例。
- 编排器执行（重放驱动）
- 引擎重放编排历史以重建状态。
- 编排器通过发布命令调度效果（活动、子编排、定时器），引擎将其持久化为新的历史事件。
- 活动分发与执行
- 引擎将活动工作项分发给 Worker。
- Worker 运行活动（可能重试，并至少一次交付）。
- Worker 以完成（结果）或失败响应；引擎追加到历史。
- 定时器与外部信号
- 引擎将定时器触发或外部事件记录作为历史条目交付。
- 编排器在下次重放时确定性地消费这些事件。
- 推进与检查点
- 每一步追加到历史日志并推进编排状态。
- 引擎保证编排状态的幂等性与恰好一次提交。
- 完成
- 编排返回输出（成功）或失败（异常详情）。
- 最终状态与输出被持久化；状态查询反映终止状态。
协议原则
- GRIEF（GRpc IntErFace）：所有 Worker/引擎与客户端/引擎通信均为 gRPC。
- 基于重放的编排：通过历史重放保证确定性。
- 活动至少一次交付：活动可能重新执行；应设计为幂等。
- 引擎中介的效果：所有非确定性/时间/IO 均通过引擎流动以保持重放安全。
文档目录
- 管理 API
详细的 Dapr gRPC 控制平面操作与负载。
- 执行 API（Task Hub 协议）
TaskHubSidecarService Worker 协议、工作项约定、结果/失败报告与排序。 - 编排生命周期
重放语义、调度、外部事件、定时器与完成。
- 活动生命周期
分发、重试、幂等、心跳语义与失败处理。
- 状态与历史
历史架构、状态快照与持久化保证。
- 版本管理
Dapr 如何处理同一工作流定义的多个版本。
1 - Workflow 协议 - 管理 API
Workflow 构建块内部的底层描述。
Workflow 管理 API
Workflow 管理 API 允许 Dapr 客户端控制 workflow 实例的生命周期。这些 API 通过标准 Dapr gRPC 端点公开，通常通过 SDK 提供。
gRPC 服务定义
管理 API 是 dapr.proto.runtime.v1 中 Dapr 服务的一部分。虽然可能存在多个版本（Alpha1、Beta1），以下描述的是当前的实现逻辑。
StartWorkflow
启动一个新的 workflow 实例。
请求（StartWorkflowRequest）：
| 字段 | 类型 | 描述 |
|---|
instance_id | string | 可选。workflow 实例的唯一标识符。如果未提供，Dapr 将生成一个随机的 UUID。 |
workflow_component | string | 要使用的 workflow 组件的名称。目前，Dapr 使用内置引擎。 |
workflow_name | string | 要执行的 workflow 定义的名称。 |
options | map<string, string> | 可选。组件特定选项。 |
input | bytes | 可选。workflow 实例的输入数据，通常是 JSON 序列化字符串。 |
响应（StartWorkflowResponse）：
| 字段 | 类型 | 描述 |
|---|
instance_id | string | 已启动的 workflow 实例的 ID。 |
GetWorkflow
检索 workflow 实例的当前状态和元数据。
请求（GetWorkflowRequest）：
| 字段 | 类型 | 描述 |
|---|
instance_id | string | 要查询的 workflow 实例的 ID。 |
workflow_component | string | workflow 组件的名称。 |
响应（GetWorkflowResponse）：
| 字段 | 类型 | 描述 |
|---|
instance_id | string | workflow 实例的 ID。 |
workflow_name | string | workflow 的名称。 |
created_at | Timestamp | 实例创建的时间。 |
last_updated_at | Timestamp | 实例最后更新的时间。 |
runtime_status | string | 状态（例如 RUNNING、COMPLETED、FAILED、TERMINATED、PENDING）。 |
properties | map<string, string> | 额外的组件特定元数据。 |
TerminateWorkflow
强制终止正在运行的 workflow 实例。
请求（TerminateWorkflowRequest）：
| 字段 | 类型 | 描述 |
|---|
instance_id | string | 要终止的 workflow 实例的 ID。 |
workflow_component | string | workflow 组件的名称。 |
RaiseEventWorkflow
向正在运行的 workflow 实例发送事件。
请求（RaiseEventWorkflowRequest）：
| 字段 | 类型 | 描述 |
|---|
instance_id | string | workflow 实例的 ID。 |
workflow_component | string | workflow 组件的名称。 |
event_name | string | 要引发的事件的名称。 |
event_data | bytes | 与事件关联的数据。 |
PauseWorkflow & ResumeWorkflow
暂停或恢复 workflow 实例。
请求（PauseWorkflowRequest / ResumeWorkflowRequest）：
| 字段 | 类型 | 描述 |
|---|
instance_id | string | workflow 实例的 ID。 |
workflow_component | string | workflow 组件的名称。 |
PurgeWorkflow
移除与 workflow 实例关联的所有状态和历史记录。这通常只能对已完成、失败或已终止的实例执行。
请求（PurgeWorkflowRequest）：
| 字段 | 类型 | 描述 |
|---|
instance_id | string | workflow 实例的 ID。 |
workflow_component | string | workflow 组件的名称。 |
ListInstanceIDs（仅限 Task Hub 协议）
检索 workflow 实例 ID 列表，可选择按状态或名称筛选。这目前是内部 Task Hub 协议的一部分，用于管理工具中的分页。
请求（ListInstanceIDsRequest）：
| 字段 | 类型 | 描述 |
|---|
page_size | int32 | 要返回的 ID 的最大数量。 |
continuation_token | string | 用于检索下一页结果的不透明 token。 |
响应（ListInstanceIDsResponse）：
| 字段 | 类型 | 描述 |
|---|
instance_ids | repeated string | 实例 ID 列表。 |
continuation_token | string | 用于下一页结果的 token。 |
继续标记和分页
continuation_token 是由 Dapr 运行时（以及底层状态存储）生成的不透明字符串。其目的是允许客户端可靠地对大量 workflow 实例进行分页，而无需一次性将所有 ID 加载到内存中。
SDK 对继续标记的要求：
- 不透明性：SDK 必须将此 token 视为黑盒。它不应尝试解析、修改或构造自己的 token。
- 状态管理：当 SDK 收到
ListInstanceIDsResponse 时，如果打算获取更多结果，应存储 continuation_token。 - 请求传播：要获取下一页，SDK 必须将从上一个响应中收到的确切
continuation_token 传递到下一个 ListInstanceIDsRequest。 - 终止：响应中的空或 null
continuation_token 表示没有更多页面可检索。
运行时行为：
运行时从状态存储的 KeysLike 操作派生此 token。由于它与底层数据库的分页机制相关联，因此该 token 可能具有过期时间或与初始请求中使用的特定查询参数（如 page_size）相关联。
实现细节
边车接收这些请求并将其转换为底层 durabletask-go 客户端的操作。例如，StartWorkflow 调用后端以创建新的业务流程实例并排队 ExecutionStarted 事件。
2 - Workflow Protocol - Execution API
Workflow 构建块内部机制的低级描述。
Workflow Execution API (Task Hub Protocol)
Workflow Execution API 是一个低级 gRPC 协议，Dapr Workflow SDK 通过它充当"Worker"。SDK 通过此协议连接 Dapr sidecar 以轮询任务并报告完成状态。
该服务名为 TaskHubSidecarService。
Service Definition (gRPC)
service TaskHubSidecarService {
rpc GetWorkItems(GetWorkItemsRequest) returns (stream WorkItem);
rpc CompleteOrchestratorTask(OrchestratorResponse) returns (CompleteBatchResponse);
rpc CompleteActivityTask(ActivityResponse) returns (CompleteBatchResponse);
// ... other management methods
}
Worker 生命周期
- 连接：SDK 打开一个到
GetWorkItems 的长连接双向流。 - 轮询：SDK 从流中接收
WorkItem 消息。 - 执行：
- 如果工作项是 Orchestration，SDK 获取并重放历史事件以确定下一步操作。
- 如果工作项是 Activity，SDK 执行活动逻辑。
- 完成：
- 对于 Orchestration，SDK 调用
CompleteOrchestratorTask 并附带要执行的操作列表。 - 对于 Activity，SDK 调用
CompleteActivityTask 并附带结果或失败信息。
gRPC Service: TaskHubSidecarService
GetWorkItems
打开流以接收 orchestration 和 activity 的工作项。
Request (GetWorkItemsRequest)：
通常为空或包含 worker 元数据。
Response (stream WorkItem)：
WorkItem 可以是以下之一：
orchestrator_item：包含某个 orchestration 的历史和新事件。activity_item：包含单个 activity 任务的详情。
CompleteOrchestratorTask
报告 orchestration 执行的结果。
Request (OrchestratorResponse)：
instance_id：workflow 实例的 ID。actions：OrchestratorAction 消息列表。custom_status：可选的用户定义状态字符串。
OrchestratorAction 类型：
ScheduleTask：调度一个新的 activity。CreateTimer：调度一个持久化计时器。CreateSubOrchestration：启动一个子 workflow。CompleteOrchestration：将 workflow 标记为完成（成功或失败）。TerminateOrchestration：强制终止实例。SendEvent：向另一个 workflow 发送事件。
CompleteActivityTask
报告 activity 执行的结果。
Request (ActivityResponse)：
instance_id：workflow 实例的 ID。task_id：activity 任务的唯一 ID。completion_token：ActivityWorkItem 中收到的 opaque token。result：activity 的序列化输出（如果成功）。failure_details：错误详情（如果失败）。
数据模型
HistoryEvent
Dapr 中的 workflow 是事件溯源的。Orchestration 的状态通过重放一系列 HistoryEvent 消息来重建。
常见事件类型：
ExecutionStarted：初始事件，包含 workflow 名称和输入。TaskScheduled：一个 activity 被调度。TaskCompleted：一个 activity 成功完成。TaskFailed：一个 activity 失败。TimerCreated：一个计时器被调度。TimerFired：一个计时器到期。OrchestrationCompleted：workflow 完成。
FailureDetails
用于报告来自 activity 或 orchestration 的错误。
error_type：标识错误类型的字符串。error_message：人类可读的错误消息。stack_trace：可选的堆栈跟踪。is_non_retriable：布尔标志。
协议细节
- 流式传输：
GetWorkItems 是服务端到客户端的流。Dapr 在工作可用时将工作推送到 SDK。 - 粘性会话：Dapr 尝试将同一实例的工作项发送到同一 worker（如果可能），但 SDK 不能依赖此特性来保证正确性。
- 确定性：SDK 必须确保 orchestration 逻辑是确定性的。在重放期间，SDK 使用
OrchestratorWorkItem 中提供的历史记录，以避免重新执行已记录的操作。
3 - Workflow Protocol - Orchestration Lifecycle
Workflow 构建块内部机制的底层描述。
编排生命周期
本文档从协议层面描述编排的生命周期，特别是 Dapr 引擎与 SDK 如何交互以可靠地执行工作流逻辑。
基于重放的执行
Dapr Workflow 使用 事件溯源 和 重放 来维护状态。Dapr 不是保存整个 worker 进程的状态（栈、变量等），而是保存已发生事件的历史记录。
重放循环
- 工作项到达：Dapr 引擎通过
GetWorkItems 流向 SDK 发送一个 OrchestratorWorkItem。该工作项包含工作流实例的完整历史记录以及任何新事件（例如，activity 完成或外部事件）。 - 重建：SDK 从头开始执行编排函数。
- 确定性执行：随着函数的执行，它会遇到"任务"（例如，调用 activity、休眠）。
- 对于每个任务，SDK 会检查提供的 History 以查看该任务是否已完成。
- 如果任务在历史记录中，SDK 会立即返回记录的结果，而无需实际重新执行任务逻辑。
- 如果任务不在历史记录中，SDK 会记录该任务需要被调度并暂停编排函数的执行（通常通过抛出特殊异常或返回待处理的 promise）。
- 报告：一旦编排函数被暂停或完成，SDK 会向 Dapr 发送
CompleteOrchestratorTask 请求。该请求包含引擎应执行的一系列 Actions（例如 ScheduleTask、CreateTimer）。 - 状态提交：Dapr 引擎接收这些操作，在状态存储中更新工作流历史记录，并调度任何请求的任务（例如，通过向 activity worker 发送工作）。
分步示例
假设一个工作流：Activity A -> Activity B。
1. 工作流启动
- 引擎：将
ExecutionStarted 事件加入队列。 - SDK：接收包含
[ExecutionStarted] 的 OrchestratorWorkItem。 - SDK：运行函数。函数调用
Activity A。 - SDK：检查历史记录。
Activity A 不在其中。 - SDK：暂停。发送包含
[ScheduleTask(Activity A)] 的 CompleteOrchestratorTask。 - 引擎：在历史记录中记录
TaskScheduled(Activity A)。
2. Activity A 完成
- 引擎：在历史记录中记录
TaskCompleted(Activity A, result="foo")。 - SDK：接收包含
[ExecutionStarted, TaskScheduled(A), TaskCompleted(A)] 的 OrchestratorWorkItem。 - SDK：从头开始运行函数。
- SDK：函数调用
Activity A。SDK 在历史记录中找到 TaskCompleted(A)。返回 "foo"。 - SDK：函数调用
Activity B。 - SDK：检查历史记录。
Activity B 不在其中。 - SDK：暂停。发送包含
[ScheduleTask(Activity B)] 的 CompleteOrchestratorTask。 - 引擎：记录
TaskScheduled(Activity B)。
3. 工作流完成
- Activity B 完成。
- SDK：接收包含 A 和 B 都已完成的历史记录。
- SDK：运行函数。A 和 B 都从历史记录返回结果。
- SDK：函数完成并返回最终结果。
- SDK：发送包含
[CompleteOrchestration(result="final")] 的 CompleteOrchestratorTask。 - 引擎：记录
OrchestrationCompleted 并将实例标记为 COMPLETED。
SDK 作者的关键要求
1. 确定性
编排函数必须是确定性的。它不能使用：
- 随机数。
- 当前日期/时间（必须使用持久化计时器或提供的
CurrentUtcDateTime）。 - 直接 IO（必须在 activity 中完成）。
- 在重放之间可能更改的全局状态。
2. 修补（运行中更新）
当工作流已经在运行时，您可能需要更新其逻辑。然而，由于工作流是基于重放的，直接更改逻辑会破坏进行中实例的确定性。
Dapr 提供了 Patching 机制（例如 ctx.IsPatched("patch-id")）来安全地引入更改：
- 逻辑分支：SDK 提供一个 API 来检查特定"补丁"是否对当前实例处于活动状态。
- 补丁记录：当执行过程中遇到补丁检查时，结果（true/false）会被记录在工作流历史中。
- 一致性：一旦补丁被记录为对实例活动（或非活动），它将在该实例的生命周期内保持如此，即使 worker 代码更改或实例被移动到另一个 worker。
- 安全性：Dapr 引擎验证重放期间遇到的补丁序列是否与历史中的序列完全匹配。如果不匹配，工作流将进入 Stalled 状态以防止数据损坏。
3. 命名版本（运行中更新）
Dapr 还提供了 命名版本控制 机制，其中 SDK 维护可用命名工作流版本的注册表。当它收到通过名称初始化新工作流的请求时，它将查询注册表以确定该名称是否匹配与指定工作流名称不同的工作流版本，并负责将请求重定向到预期的"最新"版本。
- 逻辑分支：SDK 提供一个 API 来为给定的工作流名称注册不同的版本。
- 重放一致性：运行工作流的请求可能包含一个属性，指定要执行的特定工作流名称。这确保进行中的工作流将始终使用相同的工作流版本运行，而新工作流将使用最新的可用版本。
3. 停滞状态
当引擎检测到需要手动干预或代码修复才能继续的不可恢复条件时，工作流实例进入 STALLED 状态。常见原因包括：
- 补丁不匹配：当前代码的补丁逻辑与实例的历史记录相矛盾。
- 执行错误：发生了无法通过重试处理的致命错误。
当停滞时，实例停止执行但保留在系统中。一旦根本问题得到解决（例如，部署了正确的代码版本），实例就可以恢复或将在下一个事件时自动恢复。
4. 历史管理
SDK 必须高效地搜索历史记录。通常，这是通过维护执行期间遇到的任务计数器并将它们与历史中的事件序列进行匹配来完成的。
5. 优雅暂停
SDK 需要一种机制，在任务已调度但尚未完成时停止编排函数的执行，同时不丢失稍后重新启动它的能力。
4 - Workflow Protocol - Activity Lifecycle
Workflow 构建块内部机制的低层级描述。
Activity 生命周期
Activity 是 Dapr Workflow 中的基本工作单元。与编排不同，Activity 不会重放，也不需要具有确定性。它们每次"调度"仅执行一次（尽管可能会发生重试）。
执行流程
- 调度：编排通过向 Dapr 引擎发送
ScheduleTask 操作来请求一个 Activity。 - 工作项分发：Dapr 引擎将 Activity 任务加入队列。当 Activity worker（SDK）可用时，引擎通过
GetWorkItems 流发送一个 ActivityWorkItem。 - 执行：SDK 接收
ActivityWorkItem，其中包含：name：要执行的 Activity 的名称。input：Activity 的输入数据。instance_id：调度该 Activity 的工作流实例的 ID。task_id：此特定 Activity 执行的唯一标识符。task_execution_id：此特定 Activity 的特定尝试的唯一标识符。这对于在 Activity 逻辑中实现幂等性很有用。completion_token：一个不透明 token，用于将响应与此特定工作项关联。
- 报告：Activity 逻辑完成后，SDK 向 Dapr 发送
CompleteActivityTask 请求。- 成功：SDK 在
result 字段中提供序列化输出。 - 失败：SDK 提供
failure_details（错误消息、类型、堆栈跟踪）。
Task Execution ID
task_execution_id（也称为 Task Execution Key）是一个唯一的、运行时生成的字符串（通常是 UUID），用于标识执行 Activity 任务的特定尝试。
对 SDK 的重要性
虽然 Workflow SDK 在工作项之间通常是无状态的，但 task_execution_id 为 Activity Worker 提供了关键的上下文：
- 分布式幂等性：如果 Activity 执行副作用（例如，扣款信用卡），它应该使用
task_execution_id 作为幂等性键。 - 区分重试：与
task_id（在工作流中特定步骤保持不变）不同，task_execution_id 在引擎每次重试 Activity 时都会变化（例如，由于超时或 worker 崩溃）。 - Zombie 检测：如果 Activity worker 花费时间过长，引擎使其超时并在另一个 worker 上重试，原始 worker 可能最终会完成。通过与持久存储或外部 API 核对
task_execution_id，worker 可以确定它是否是一个不再需要其结果的"zombie"。
SDK 实现指南：
- 向用户公开：SDK 必须将
task_execution_id 公开给 Activity 实现逻辑（例如，通过 ActivityContext）。 - 不要缓存：SDK 不应尝试跨不同工作项缓存或重用此 ID。
- 不透明使用：SDK 应将该值视为不透明字符串。它由 Dapr 边车在分发 Activity 时生成，不是 SDK 需要创建或解析的东西。
Completion Tokens
completion_token 是由 Dapr 运行时生成的不透明字符串，并作为 ActivityWorkItem 的一部分传递给 SDK。
目的和意图
- 响应关联：边车使用
completion_token 将 ActivityResponse（来自 CompleteActivityTask）可靠地匹配到它分发的原始任务。 - 无状态跟踪：它允许边车在接收完成时保持无状态或最小化状态查找，因为 token 包含（或指向）必要的上下文（实例 ID、任务 ID 等）。
- Zombie 防止：如果 Activity 超时并被重试，新尝试将具有不同的
completion_token。如果原始"zombie" worker 最终使用旧 token 响应，边车可以轻松识别并忽略延迟的响应。
SDK 实现指南
- 捕获：SDK 必须从传入的
ActivityWorkItem 中捕获 completion_token。 - 传播：SDK 必须在通过
CompleteActivityTask 发送的 ActivityResponse 中包含完全相同的 completion_token。 - 不透明性：SDK 必须将 token 视为黑盒。它不应尝试解析、修改或构造自己的 token。
- 存储：在 Activity 执行期间，SDK 必须将此 token 保存在内存中（例如，在
ActivityContext 中）。
Task Activity IDs
在 Dapr 运行时中（特别是使用 Actors 后端时），Activity 被表示为 actors。每个 Activity 执行都有一个唯一的 Task Activity ID（也称为 Activity Actor ID）。
ID 遵循特定模式：
{workflowInstanceID}::{taskID}::{generation}
- workflowInstanceID：调度该 Activity 的工作流实例的唯一 ID。
- taskID：工作流执行中任务的序列号（例如，0、1、2…）。
- generation：如果工作流被重启或"继续作为新工作流"，该计数器会递增。
这个唯一的 ID 确保 Activity 执行被隔离，并且可以在重试和重启期间可靠地跟踪。
重试
Dapr 根据编排中定义的策略处理 Activity 重试（如果 SDK 支持在 ScheduleTask 操作中定义重试策略）。如果 Activity 失败并且存在重试策略，引擎将在指定的延迟后重新将 Activity 任务加入队列。
从 Activity worker 的角度来看，重试只是一个具有相同名称和输入的新 ActivityWorkItem，但可能具有不同的 task_id（或相同的，取决于后端实现）。
幂等性
因为 Activity 可能被执行多次（例如，如果 worker 在执行之后但在报告完成之前崩溃），所以建议 Activity 逻辑尽可能具有幂等性。
与工作流的比较
| 功能 | 编排 | Activity |
|---|
| 执行方式 | 基于重放（确定性） | 直接执行 |
| 状态 | 通过历史事件管理 | 无内部工作流状态 |
| 副作用 | 禁止（必须使用 Activity） | 允许（IO、数据库等） |
| 生命周期 | 可以长时间运行（天/月） | 通常短期 |
| 连接性 | 通过 GetWorkItems 连接 | 通过 GetWorkItems 连接 |
5 - Workflow Protocol - State & History
Workflow 构建块内部机制的低层描述。
状态和历史管理
Dapr Workflows 采用事件溯源模式，这意味着工作流的状态是从一系列事件中推导出来的。本文档描述了 Dapr 如何存储和管理这些历史记录和状态。
后端存储：Dapr Actors
默认情况下，Dapr Workflow 引擎使用 Dapr Actors 作为其存储后端。每个工作流实例都映射到一个唯一的 actor 实例。这提供了以下特性：
- 并发控制：Actors 确保在任意时刻只有一个操作在处理工作流实例。
- 可靠性：Actor 状态会持久化到配置好的 Dapr 状态存储中。
- 定时器：Dapr Actors 提供持久化提醒，用于实现工作流定时器。
工作流状态架构
工作流实例（actor）的状态由以下几个组件构成：
1. 元数据
存储有关该实例的高级信息：
instance_id：工作流的唯一 ID。name：工作流的名称。status：当前运行时状态（如 Running、Completed、Stalled 等）。version：工作流版本的名称及任何活跃的补丁。created_at：创建时间戳。last_updated_at：最后活动时间戳。input：原始输入数据。output：最终输出数据（如已完成的）。
2. 历史记录
一系列 HistoryEvent 对象，记录工作流中发生的所有事件。为了优化大型历史记录，Dapr 通常将历史事件以分块或单独键值的形式存储在状态存储中：
- 键格式：
wf-history-<instance_id>-<index> - 事件内容：序列化的 protobuf 消息，包含事件类型、时间戳以及类型特定的数据（如
TaskScheduled、TaskCompleted）。
3. 收件箱（待处理事件）
已发生但尚未被编排器处理（重放）的事件集合。包括：
- 向工作流引发的外部事件。
- 已完成的 activity 结果。
- 已触发的定时器。
当编排器下次运行时，它会"排空"收件箱，将这些事件移入历史记录，然后重放逻辑。
重放与状态重建
当 Worker（SDK）收到工作项时，Dapr 提供历史事件。SDK 通过按顺序重放这些事件来重建编排的内部状态（例如本地变量、当前执行点）。
确定性原则与历史记录
历史记录是"真相的来源"。如果编排代码以非确定性的方式发生变化（例如在现有代码中间添加新的 activity 调用），重放将失败，因为代码的请求将与记录的历史记录不匹配。
清除状态
当清除工作流时，Dapr 会从状态存储中删除元数据及所有相关的历史事件键。这通常是为了在完成工作流后进行清理。
6 - 工作流协议 - 版本控制
工作流构建块内部实现的底层描述。
工作流版本控制
Dapr 工作流支持工作流定义的版本控制，允许你在现有实例继续运行其原始逻辑的同时更新工作流逻辑。
命名工作流版本控制
向 Dapr 引擎注册工作流时，你可以提供版本名称。这允许同一工作流的多个版本共存。
- 默认版本：一个工作流版本可被标记为默认版本。如果客户端仅通过名称启动工作流而不指定版本，则使用默认版本。
- 特定版本：客户端可以在启动新实例时请求工作流的特定版本。
注册 API
SDK 使用其任务注册表中的 AddVersionedOrchestrator（或类似）方法来注册版本化工作流。
// 示例（内部注册表 API）
registry.AddVersionedOrchestrator("MyWorkflow", "v2", true, MyWorkflowV2)
registry.AddVersionedOrchestrator("MyWorkflow", "v1", false, MyWorkflowV1)
边车版本控制
Dapr 边车会跟踪实例正在运行的工作流命名版本，以及在该工作流执行过程中已应用的补丁列表。此信息存储在工作流历史记录中，位于 OrchestratorStarted 事件的 version 字段内。
message OrchestrationVersion {
string name = 1;
repeated string patches = 2;
}
当实例恢复时（例如，在某个活动完成后），Dapr 引擎负责处理因客户端版本不匹配而导致的停滞。它通过监控 placement 表的变化来实现这一点，这表明有新的 SDK 客户端已连接（可能代表应用程序的较新副本实例），并调度重放最后一个事件以尝试重试该操作。这一次，如果 SDK 能够成功完成任务，运行时将把工作流状态从 Stalled 更改回 Running。