Rill Flow 执行机制解析：DAG 调度、异步回调与失败重试

一个视频上传后，需要探测媒体信息、拆分片段、并行转码，最后合并结果。把这些接口依次调用并不难；困难出现在第三个分片失败、其他分片已经完成时：应该重跑整个流程，还是只重试失败节点？如果执行器重复回调，合并任务会不会被触发两次？

Rill Flow 用 DAG 描述任务依赖，由引擎管理执行状态，再把具体计算交给执行器。理解它的关键，是沿着一次执行观察：谁决定节点可以运行，谁记录完成，以及失败以后由谁再次推进。

本文以公开仓库的 5cded0b 版本 为依据，讨论普通非流式输入、未启用关键路径提前完成的流程。视频处理链路是教学案例；代码与配置经过静态核对，未在本文中启动 Rill Flow 执行。已有实现与额外设计建议会分别说明。

从一个并行视频处理任务说起

先把业务依赖画出来：

媒体探测 → 分片
            │
            ├→ 转码分片 0 ─┐
            ├→ 转码分片 1 ─┼→ 合并 → 结果
            └→ 转码分片 2 ─┘

每个转码任务都依赖分片结果，但转码任务之间没有先后关系。合并必须等待所需分片就绪。DAG 的价值在于把这两个条件显式表达出来，让并行机会和等待关系都可以被引擎检查。

还需要区分“流程定义”和“执行实例”：定义描述有哪些步骤，同一份定义可以处理许多视频；实例则拥有自己的 executionId、节点状态与运行上下文。排查某个视频卡住时，应先定位执行实例，再查具体节点和调用记录，不能只看配置图是否正确。

用 YAML 描述依赖与数据传递

Rill Flow 的官方并行异步示例 展示了 foreach、异步函数节点和结果汇总。下面沿用其结构，把业务对象换成视频分片，只保留“并行转码 → 合并”这段链路：

version: 0.0.1
type: flow
workspace: videoDemo
dagName: parallelTranscode
tasks:
  - category: foreach
    name: transcodeSegments
    next: merge
    inputMappings:
      - source: $.context.segments
        target: $.input.segments
    iterationMapping:
      collection: $.input.segments
      item: segment
    tasks:
      - category: function
        name: transcode
        pattern: task_async
        resourceName: https://transcoder.invalid/transcode
        tolerance: false
        inputMappings:
          - source: $.context.segment
            target: $.input.segment
        outputMappings:
          - source: $.output.result_url
            target: $.context.result_url
    outputMappings:
      - source: $.output.sub_context.[*].result_url
        target: $.context.transcoded_segments
  - category: function
    name: merge
    pattern: task_sync
    resourceName: https://transcoder.invalid/merge
    tolerance: false
    inputMappings:
      - source: $.context.transcoded_segments
        target: $.input.files

这是教学配置：初始上下文需要提供 segments，.invalid 地址是占位符，转码服务也需要实现异步回调协议，因此不能直接复制上线。要运行官方案例，应使用该版本的部署说明及配套示例执行器；官方案例处理的是数值，不会替你完成视频转码。

这段配置同时定义了控制关系和数据关系。next: merge 表示 foreach 节点之后执行合并；inputMappings 把上下文中的数据放进当前节点输入；outputMappings 则把节点输出写回上下文。依赖满足和数据准备好，是必须同时维护的两个条件。

foreach 会根据集合创建多组子任务，并为各组建立子上下文。这里的 segment 来自当前迭代项，子任务的 result_url 写入所属组的上下文，再由父节点从 sub_context 汇总。不能把所有分片都理解成向一个全局字段并发覆盖。实际合并协议还应携带分片序号，并按序号排序、检查缺片；这些是业务约束，示例中的 URL 列表不足以独立保证。

一次工作流执行如何推进

从描述文本到可运行节点

DAGStringParser 负责把描述解析为 DAG 模型；FlowDAGValidator 进一步检查命名、任务引用和路径中的环等约束。YAML 能被解析，只能证明格式成立，不能证明图中的后继节点存在。正式接入还要验证执行器地址、参数及响应协议。

执行时，引擎围绕实例维护图信息、任务信息和上下文。DAGTraversal 读取状态，借助 DAGWalkHelper 找到可以启动的节点。在本文限定的普通模式下，候选节点需要处于 NOT_STARTED，且所有依赖均已成功或被跳过；没有前置依赖的节点可以作为入口。已经失败的前置节点，并不因为“执行结束”就自动满足依赖。

为什么先保存 READY

DAGTraversal.submitTraversal 在实例对应的锁内推进遍历。选出任务后，普通执行路径先把节点设成 READY 并保存，再交给 DAGOperations.runTasks。另一个完成事件触发遍历时，节点已不再是 NOT_STARTED，可以避免再次被当作新任务选中。

随后，DAGOperations 按任务类别选择 runner：函数节点进入 FunctionTaskRunner，foreach 则由对应 runner 展开子任务组。执行结果可能要求等待回调、安排重试，也可能已经完成；完成时再次提交遍历，寻找后继节点，而不是一开始就算好顺序后一路调用到底。

但状态保存与外部请求之间仍有时间窗口。若进程在保存后、请求发出前退出，单靠 READY 标记无法证明任务会自动恢复。锁与状态更新解释了正常并发推进时如何减少重复，不能直接推导出跨进程故障下的完整执行保证。

状态表要对应真实实现

该版本的 TaskStatus 定义如下，不能直接用一张通用工作流状态表替代：

这些状态不是一条严格向前的直线。例如重试会重新进入 READY；KEY_SUCCEED 也不能在普通汇总场景里无条件当作全部完成。阅读状态判断时，要一起看它所属的运行模式。

异步任务怎样完成与汇合

task_sync 根据当前调用返回的结果判断完成状态；task_async 在派发返回后保持 RUNNING，等待后续完成通知。执行器返回“已受理”，只说明任务已进入它的处理流程，不代表目标分片已经生成。

以一个异步转码节点为例，正常路径可以缩写为：

引擎 → 执行器：派发分片任务
执行器 → 引擎：返回受理结果
引擎：节点保持 RUNNING
执行器：完成计算，发送完成回调
引擎：校验节点，写输出，保存完成状态
引擎：再次遍历，推进父节点或后继节点

官方示例执行器从 X-Callback-Url 获取引擎回调地址，并在计算结束后向该地址提交结果。配置中另一个名为 callback 的节点，则是普通的后续业务函数，用于继续处理结果。二者名称相似，但一个是完成通知，一个是 DAG 中的任务，不能混为一谈。

这里还有一个值得注意的写入顺序：FunctionTaskRunner.handleNormalCallback 先完成输出映射并保存上下文，再保存任务状态。否则，另一条并行分支的完成事件可能让后继节点看到“依赖已成功”，却读不到刚产生的输出。这体现的是数据可见性与状态推进之间的配合，不代表这些写入天然构成跨系统事务。

对于本文未开启容错跳过和关键路径模式的案例，各组转码都成功后，父 foreach 汇总子上下文，再让 merge 获得运行条件。某组失败不会自动变成有效的视频结果。合并节点仍需检查产物是否存在、分片数是否正确，以及编码参数是否兼容。

重复调度与失败恢复的边界

完成通知怎样防止重复推进

函数节点的完成处理在任务对应的锁内重新读取状态。finishActionValidateTaskInfo 会拒绝对 NOT_STARTED、READY 节点提交完成，也会拒绝已经处于终态的节点再次完成。重复通知进入这条路径时，会得到非法状态异常，而不是再次正常写入完成状态。

这仍然不能保证外部业务只执行一次：请求可能已被执行器接收，但引擎没有收到响应；再次派发时，执行器可能重复生成文件。业务侧应设计稳定的幂等键，例如执行实例、分片标识和转码参数版本的组合，并持久化处理结果。这个建议需要在执行器中实现，不能仅依赖调度锁。

同样，终态检查不等于对每次尝试都有隔离。超时重试后，旧请求的结果仍可能迟到；应结合调用标识和结果版本检查，验证它是否属于当前有效尝试，不能从一次状态检查推断所有迟到回调都安全。

失败以后如何决定重试

SimpleRetryPolicy 先要求状态为 FAILED，再根据已记录的调用次数与 maxRetryTimes 决定是否重试。如果配置了 conditions，还需要至少一个重试条件与输出匹配；引擎不会自行理解“参数错误不值得重试”，需要业务把错误信息转换成可判断的数据。

间隔按 intervalInSeconds × multiplier^invokeTimes 计算，并转为整数秒。若当前调用次数为 1，基础间隔为 3 秒、倍率为 2，算出的间隔就是 6 秒，不能按“第一次一定等基础间隔”来理解。该策略没有加入随机抖动或间隔上限，写配置时需要留意这一点。

需要重试时，runner 保存 READY，返回重试标记和间隔，由 DAGOperations 安排立即或延迟运行。仓库中的 FunctionRetryTest 包含“最多重试 2 次、连续派发失败时共调用 3 次”的断言；这是测试源码提供的行为证据，本文没有实际运行该测试。

重试不再适用时，失败结果会进入普通完成处理；对允许容错的函数节点，tolerance: true 可使失败变成 SKIPPED。后继节点可能因此继续执行，但跳过不会产生缺失的业务结果。对于要求所有分片齐全的合并流程，不应通过容错配置掩盖缺片。

用故障场景检查保证范围

排障记录也应围绕这些问题组织：关联执行实例、子任务组、调用次数、状态变化、错误原因和产物地址。保存输入输出摘要有助于定位故障，但查看历史轨迹、重新执行和确定性重放是不同能力，不能统一写成“支持回放”。

扩展到 LLM 工作流时，哪些机制可以复用

把视频分片换成待处理文档，把转码换成信息提取或模型生成，依赖管理、并行分组、上下文传递和异步完成通知仍然有用。例如分别提取多个文档的信息，再汇总为答案，就能沿用“展开子任务 → 各自执行 → 收集结果”的结构。

变化主要在节点成功的定义。转码服务可以检查目标文件是否生成；模型返回 HTTP 200，却可能缺少字段、引用不存在的材料，或者给出结构正确但内容错误的答案。引擎已有成功条件、失败条件和重试条件的判断入口，但质量规则需要由应用提供。JSON 校验只能解决协议问题，事实性与业务质量还需要额外评估。

一种扩展方案是让模型执行器先把结果归一化，明确区分服务不可用、格式错误和质量不达标，再交给工作流选择后续动作。临时限流可以按预算退避；格式错误可以有限次数地修复；证据不足则应补充检索或结束流程。不要让所有错误都退化成重新调用同一个模型。

成本也应成为业务状态的一部分。对整次执行设置最大模型调用次数和费用预算，记录模型版本、提示词版本及校验失败原因。这样才能区分“某个节点重试次数未超限”与“整条链路已经消耗过多”。这些是需要补充的应用设计，并非本文源码已经证明存在的内置功能。

还要区分 DAG 内的有限集合展开与开放式智能体循环。foreach 可以处理已给定的集合，但不能据此推导出任意循环都会被引擎自动终止。若要根据模型结果反复检索、修改计划，应另外明确轮次上限、退出条件和状态归属，再决定如何把这段控制逻辑接入工作流。

适用边界与源码索引

当流程出现并行依赖、跨服务异步执行，以及需要单独定位和恢复的节点时，Rill Flow 的状态模型和执行入口值得研究。只有几个短小的顺序调用时，引入工作流也会增加配置、存储和运行维护成本，应先判断这些能力是否真的需要。

读源码可以从以下入口串起一条路径，所有链接均固定到本文版本：

• 解析与校验：DAGStringParser、FlowDAGValidator。
• 可运行条件：DAGWalkHelper、图推进：DAGTraversal。
• 任务执行与再次调度：DAGOperations。
• 函数派发、结果映射与完成处理：FunctionTaskRunner、完成状态校验：AbstractTaskRunner。
• 集合展开：ForeachTaskRunner、重试行为的测试源码：FunctionRetryTest。
• 并行异步配置与配套执行器：对照阅读请求受理、后台计算和完成回调。