RandallFlare Flow：一个可视化工作流引擎的前后端算法#

Flow 是 RandallFlare 边缘平台上的 n8n 式可视化工作流编排器：在画布上拖拽节点， 每个节点驱动一个平台原语（Worker / KV / D1 / R2 / Queue / Workflow / Email / Analytics …），节点间连边传数据，支持分支、循环、子流程、跨节点表达式、 同步 API、失败告警与版本回滚。

这篇文章讲清楚它背后的两组核心算法：后端的 DAG 执行引擎，和 前端的节点自动整理（布局）算法。

一、Flow 是什么#

传统的 Serverless 平台里，把几个函数串成一个业务流程，你得手写胶水代码：调用 A、 拿结果、调 B、判断、写数据库、发邮件……Flow 把这件事可视化了。

一个 Flow 本质是一张有向图（DAG，允许受控回边做循环）：

• 节点（node） = 一次操作（跑一个 Worker、查一次 D1、发一封邮件……）
• 边（edge） = 数据流向 + 执行依赖
• 触发器（trigger） = 手动 / crontab / webhook

它不是又造一个运行时——执行完全复用 RandallFlare 已有的持久化基础设施 （轮询 + 租约 + step 缓存，和 Workflow 引擎同源），只是把「图的形状」存成 JSON， 在运行时解释执行。

---

二、整体架构#

几个关键设计点：

---

三、数据模型#

model EdgeFlow {
  id          String  @id @default(cuid())
  ownerId     String
  graph       Json    // { nodes: [...], edges: [...] } —— React Flow 序列化
  triggerType String  // manual | cron | webhook
  cronExpr    String?
  alertWebhook String?
  maxConcurrentRuns Int @default(0)
  runs        EdgeFlowRun[]
  versions    EdgeFlowVersion[]
}

model EdgeFlowRun {
  id               String   @id @default(cuid())
  status           String   // PENDING | RUNNING | COMPLETE | FAILED | CANCELED
  input            Json?
  output           Json?
  graphSnapshot    Json     // 冻结触发时的 DAG —— 运行中改画布不影响在飞的 run
  sleepUntil       DateTime @default(now())   // 调度到期时间
  advanceLockUntil DateTime?                  // 乐观推进租约
  steps            EdgeFlowRunStep[]
  @@index([status, sleepUntil])
}

model EdgeFlowRunStep {
  runId     String
  nodeId    String
  iteration Int    @default(0)   // 循环迭代号
  seq       Int
  status    String // PENDING | OK | FAILED | SKIPPED
  input     Json?
  output    Json?
  @@unique([runId, nodeId, iteration])
}

EdgeFlowRunStep 的 @@unique([runId, nodeId, iteration]) 是整个持久化模型的地基： 每个节点每次迭代恰好一行，所以「推进一步」这个动作是幂等、可从数据库重建的—— 进程崩了重放不会重复执行。

---

四、后端：DAG 执行引擎#

4.1 图的表示#

记一张 Flow 图为 $G = (V, E)$：

• $V$：节点集合，每个节点 $v$ 有类型 $\tau(v)$（trigger / worker / branch / loop / …）
• $E$：有向边集合，边 $e = (u \to v)$ 带一个源句柄 $h(e)$（普通边为 null， 分支边为 true/false，循环边为 each/done，错误边为 error）

执行的目标：给定触发输入 $I_{\text{run}}$，按依赖顺序执行每个节点，把上游输出喂给下游， 最终产出终端节点的输出。

4.2 边状态与就绪判定#

引擎不做全局拓扑排序，而是增量地从已完成的 step 推导每条边的状态。设 $\sigma(v)$ 为节点 $v$ 的执行状态（未完成 / OK / FAILED / SKIPPED），并对分支类节点记其 「取用句柄」$\text{taken}(v)$。则一条边 $e=(u\to v)$ 的状态：

$$\text{state}(e) = \begin{cases} \textbf{pending} & \sigma(u) = \bot \quad(\text{源未完成}) \\[4pt] \textbf{dead} & \sigma(u) \in \{\text{SKIPPED}, \text{FAILED}\} \\[4pt] \textbf{dead} & \text{taken}(u) \neq \varnothing \ \land\ h(e) \neq \text{taken}(u) \\[4pt] \textbf{live} & \text{其它} \end{cases}$$

第三条是分支剪枝：一个 branch 节点算出 taken = "false" 后，它的 true 边 立刻变 dead，false 边变 live。loop 节点执行完返回 taken = "done"，于是 each 边 dead、done 边 live——保证循环体不会在主遍历里被重复触发。

有了边状态，节点 $v$ 就绪（ready） 当且仅当它尚未完成，且：

$$\text{ready}(v) \iff \sigma(v)=\bot \ \land\ \Big[\ \text{in}(v)=\varnothing \ \lor\ \big(\underbrace{\forall e\in\text{in}(v):\ \text{state}(e)\neq\textbf{pending}}_{\text{所有入边已定}}\ \land\ \underbrace{\exists e\in\text{in}(v):\ \text{state}(e)=\textbf{live}}_{\text{至少一条活}}\big)\ \Big]$$

对偶地，节点 $v$ 判死（skipped）——即所有入边已定但没有一条活：

$$\text{dead}(v) \iff \sigma(v)=\bot \ \land\ \text{in}(v)\neq\varnothing \ \land\ \big(\forall e:\ \text{state}(e)\neq\textbf{pending}\big) \ \land\ \big(\nexists e:\ \text{state}(e)=\textbf{live}\big)$$

判死的节点会被标 SKIPPED，从而它的出边全部变 dead，沿图传播下去（分支未走的整条子图自动熄灭）。

真实代码里的 edgeState：

function edgeState(edge: GEdge, steps: Map<string, StepState>):  "live" | "dead" | "pending" {  const s = steps.get(edge.source);  if (!s) return "pending";  if (s.status === "SKIPPED" || s.status === "FAILED") return "dead";  if (s.taken !== undefined) {    // 分支：只有被取用的句柄是 live    return edge.sourceHandle === s.taken ? "live" : "dead";  }  return "live";}

4.3 波次遍历算法#

advanceRun 一次「推进」把尽可能多的就绪节点跑掉（受预算 $B$ 约束），流程：

关键性质——为什么不会死循环也不会漏：

• 每次内层循环要么标一个死节点（有限，每节点最多一次），要么执行一个就绪节点 （受预算 $B$ 上限），要么因「无就绪且无进展」跳出。故内层循环必然终止。
• 图里若含环（用户画了回边），环上节点永远不满足「所有入边已定」→ 一直 pending → 最终没有就绪节点、也标不了死 → 跳出并落 COMPLETE（尽力而为，不 hang）。
• 超过预算 $B$ 且仍有 pending → 落 PENDING + sleepUntil = now，下次轮询继续， 从 step 无损重建。

4.4 节点输入的合并#

一个就绪节点 $v$ 的输入，取它活入边的源节点输出。设 $L(v) = \{ (u\to v)\in\text{in}(v) : \text{state}=\textbf{live}\}$，则：

$$\text{input}(v) = \begin{cases} I_{\text{run}} & \text{in}(v) = \varnothing \quad(\text{根节点用触发输入}) \\[4pt] \text{out}(u) & L(v) = \{(u\to v)\} \quad(\text{单一上游，透传}) \\[4pt] \big\{\, u \mapsto \text{out}(u) \ \big|\ (u\to v)\in L(v) \,\big\} & |L(v)| > 1 \quad(\text{多上游，按源 id 归并成对象}) \end{cases}$$

4.5 分支（branch）#

branch 节点对输入按配置求值（路径 + 运算符 + 值），得到 true/false， 作为 taken 写进 step。它透传输入（output = input），下游按 §4.2 的第三条被剪枝。

4.6 循环（loop）—— 内联子图执行#

循环是最有意思的一块。loop 节点有两个出口句柄：each（循环体）和 done（结束后）。

循环体边界用可达性定义。设 $\text{reach}(S)$ 为从节点集 $S$ 沿边前向可达的节点集， 则循环体：

$$\text{Body} = \text{reach}\big(\text{targets}(\text{each})\big) \ \setminus\ \Big(\{\text{loop}\}\ \cup\ \text{reach}\big(\text{targets}(\text{done})\big)\Big)$$

即：从 each 出口可达、但排除循环节点自身与 done 侧可达的所有节点。

执行时，先决定迭代项集合 $\text{items}$：

$$\text{items} = \begin{cases} [\,0, 1, \dots, n-1\,],\quad n=\min(\text{count},\ \text{max}) & \text{mode}=\texttt{count} \\[4pt] \text{arr}[\,0:\text{max}\,],\quad \text{arr}=\text{resolve}(\text{itemsPath}, \text{input}) & \text{mode}=\texttt{collection} \end{cases}$$

然后对每个 item 同步跑一遍循环体子图（body 内部同样跑一次微型的波次遍历）， 第 $i$ 次迭代的 step 用 iteration = i+1 记录（主图用 iteration = 0，天然不冲突）：

$$\text{out}(\text{loop}) = \big[\ \text{runBody}(\text{item}_i)\ \big]_{i=0}^{|\text{items}|-1}$$

循环节点执行完返回 taken = "done"，主遍历里 each 边熄灭、done 边点亮， 循环体不会被主图重复执行。嵌套循环（体内又有 loop）在当前实现里被跳过（有界性保护）。

4.7 子流程（subflow）#

subflow 节点把另一张 Flow 图当模块内联执行：拿本节点输入当子图的触发输入， 用全新的、隔离的上下文跑一遍子图的波次遍历，返回其终端输出。带 owner 校验（防止伪造图调用他人流程）和深度守卫（depth ≥ 3 直接失败， 防环）。

4.8 运行上下文与表达式（跨节点数据）#

这是从「玩具」到「企业级」的分水岭。MVP 里一个节点只能拿到直接上游的输出； 企业级流程需要任意节点读任意先前节点的数据。

解法：维护一份运行上下文 —— 每个已完成节点的输出，按 id 和 label 双索引：

$$\mathcal{C} = \big\{\ \text{id}(u) \mapsto \text{out}(u),\ \ \text{label}(u) \mapsto \text{out}(u)\ \ \big|\ \sigma(u)=\text{OK}\ \big\}$$

节点的任意字符串字段（url / path / headers / body / KV key / D1 sql / …）都可内嵌 {{ … }} 表达式，用 JSONata 对上下文求值：

为什么用 JSONata 而不是 eval 任意 JS？因为执行跑在多租户的共享控制面上， new Function / vm 都不是真正的沙箱，能逃逸拿到进程/文件系统。JSONata 是 声明式的、无任意代码执行，安全可控。

数据流示意（一个「查用户 → 分支 → 发 VIP 邮件」的例子，注意 Email 节点直接引用了 非相邻的「查用户」输出）：

resolveTemplate 的核心（整字段单 {{ }} 保留原始类型，否则字符串插值）：

async function resolveTemplate(tpl: string, ctx: ExprContext): Promise<Json> {  if (!tpl.includes("{{")) return tpl;  const whole = tpl.match(/^\s*\{\{([\s\S]+)\}\}\s*$/);  if (whole) return evalExpr(whole[1].trim(), ctx);   // 整字段 → 原始值  // 否则逐个 {{ }} 求值、字符串拼接  let out = "", last = 0;  const re = /\{\{([\s\S]+?)\}\}/g;  for (let m; (m = re.exec(tpl)); ) {    out += tpl.slice(last, m.index);    const v = await evalExpr(m[1].trim(), ctx);    out += typeof v === "string" ? v : v == null ? "" : JSON.stringify(v);    last = re.lastIndex;  }  return out + tpl.slice(last);}

4.9 持久化：轮询 + 租约 + step 缓存#

一次 run 的生命周期是个状态机，靠 sleepUntil（何时该推进）和 advanceLockUntil （乐观推进租约）驱动：

抢占用条件更新做原子租约——只有 updateMany 影响行数为 1 的那个推进者才真正执行， 因此同步 webhook 的内联推进与后台轮询推进永不打架：

const claim = await prisma.edgeFlowRun.updateMany({  where: { id: runId, status: "PENDING", sleepUntil: { lte: new Date() } },  data:  { status: "RUNNING", advanceLockUntil: new Date(Date.now() + LEASE_MS) },});if (claim.count === 1) { /* 只有我抢到，才推进 */ }

崩溃恢复：租约过期（advanceLockUntil < now）的 RUNNING 被重置回 PENDING， 下次轮询重新 claim——自愈。

---

五、前端：节点整理算法（自动布局）#

5.1 问题#

用户随手拖拽，节点位置很快就乱。「⤢ tidy」按钮要把任意一张图自动排成 可读的从左到右的分层布局。这是经典的**有向图分层绘图（layered / Sugiyama-style graph drawing）**问题，我们用 dagre 求解。

5.2 Sugiyama 分层绘图#

给定 DAG $G=(V,E)$，分层绘图分三步，每步都是一个优化子问题：

① 分层（rank assignment）。给每个节点一个层号 $\ell: V \to \mathbb{Z}$，让每条边都 「向右跨至少一层」，并最小化总边长：

$$\min_{\ell} \sum_{(u,v)\in E} \big(\ell(v) - \ell(u)\big) \quad \text{s.t.} \quad \ell(v) - \ell(u) \ge 1 \ \ \forall (u,v)\in E$$

这是一个整数规划，但约束矩阵全幺模（totally unimodular），线性松弛的最优解天然 取整——dagre 用**网络单纯形（network simplex）**在近似线性时间里求最优分层。

② 层内排序（ordering）。同一层内节点的先后顺序决定边的交叉数。最小化交叉是 NP-hard，用经典的**重心 / 中位数启发式（barycenter / median heuristic）**逐层扫： 把节点 $v$ 排到它邻居位置的中位数处：

$$x_{\text{order}}(v) \leftarrow \operatorname{median}\big\{\, \text{pos}(u) : u \in N(v) \,\big\}$$

反复上下扫几轮直到收敛，交叉数显著下降。

③ 坐标分配（coordinate assignment）。在保持层与序的前提下定实际像素坐标，让边尽量 「直」——最小化相邻层对齐节点的水平偏移：

$$\min_{x} \sum_{(u,v)\in E} \omega_{uv}\,\big|\,x(u) - x(v)\,\big|$$

（$\omega_{uv}$ 是边权；穿过多层的长边会被打成虚拟节点参与，保证走线平滑。）

5.3 dagre 集成 + 坐标变换#

我们把方向设成 LR（从左到右，正好对应节点「左入右出」的句柄），跑完 dagre 后有个 关键细节：dagre 返回的是节点中心坐标，React Flow 要的是左上角坐标，需要一步平移：

$$(x,\,y)_{\text{topLeft}} = \Big(x_{\text{center}} - \tfrac{w}{2},\ \ y_{\text{center}} - \tfrac{h}{2}\Big)$$

function autoLayout(nodes: FNode[], edges: Edge[]): FNode[] {  const g = new Dagre.graphlib.Graph().setDefaultEdgeLabel(() => ({}));  g.setGraph({ rankdir: "LR", nodesep: 42, ranksep: 90 });      // 分层方向 + 间距  const dim = (n) => ({ w: n.measured?.width ?? 190, h: n.measured?.height ?? 74 });  nodes.forEach((n) => { const { w, h } = dim(n); g.setNode(n.id, { width: w, height: h }); });  edges.forEach((e) => g.setEdge(e.source, e.target));  Dagre.layout(g);                                              // ① 分层 ② 排序 ③ 定坐标  return nodes.map((n) => {    const p = g.node(n.id);    const { w, h } = dim(n);    // dagre 给中心，React Flow 要左上角    return { ...n, position: { x: p.x - w / 2, y: p.y - h / 2 } };  });}

复杂度：网络单纯形分层近似 $O(|V||E|)$（实践中远低于此），中位数排序每轮 $O(|E|)$，整体对画布规模（几十到几百节点）是毫秒级。排完顺手 fitView 动画归位。

5.4 实时运行叠加#

布局之外，前端还做实时染色：每 3 秒轮询 latestFlowRunStatus，把每个节点的最新 状态（折叠多次迭代，优先级 FAILED > PENDING > OK > SKIPPED）写回节点的 data.__status， 自定义节点据此渲染边框颜色（绿/红/黄脉冲/灰）。折叠函数：

$$\text{status}(v) = \arg\max_{s\in\text{steps}(v)} \ \text{rank}(s), \quad \text{rank}: \{\text{FAILED}{:}3,\ \text{PENDING}{:}2,\ \text{OK}{:}1,\ \text{SKIPPED}{:}0\}$$

保存时这些 __ 前缀的瞬态字段会被剥掉，不进库。

---

六、数据流全景#

把一次同步 webhook 调用（Flow-as-API）从头到尾串起来：

异步模式（不带 ?wait=1）则是：建 run 立即返回 runId，靠 agent 轮询的 after(advanceDueFlows) 在后台推进——同一套引擎，两种语义。

---

七、节点类型总览#

**圆盘（数据库/存储/队列）**的单锚点方向跟着操作走：读是输出（往下游喂结果， 下游用 {{ $node['label'] }} 取），写是输入（上游连进来，把数据写进去）。 这在视觉上把「一步操作」和「一个可查询的资源」区分开了。

---

八、一个完整例子：1 → 8 循环#

图结构（可直接导入的 JSON 略）：

执行轨迹：

• Loop count 模式产出下标 $[0,1,\dots,7]$
• 循环体 Transform 对每项算 \item + 1$→$1, 2, \dots, 8$，各记一条 iteration = i+1 的 step
• Loop 输出 $[1,2,3,4,5,6,7,8]$，taken="done"
• Result 透传该数组 → run 输出 $[1,2,3,4,5,6,7,8]$

在画布上点 ▶ run，会看到 Transform 节点亮 8 次（迭代 1–8），运行详情里逐条列出 每次的输入/输出。

---

九、小结#

Flow 的两组算法，各自解决一个清晰的问题：

• 后端：不做全局拓扑排序，而是用边状态 + 就绪判定做增量的波次遍历——天然支持 分支剪枝、循环内联、跨节点上下文，且每一步都是可从数据库无损重建的幂等操作， 配合条件更新的乐观租约，得到崩溃自愈、不重复、可同步可异步的持久化执行。

• 前端：把「整理乱图」归约成经典的分层绘图，用 dagre 的网络单纯形分层 + 中位数排序 + 坐标最优化求解，再做一步中心→左上角的坐标变换接进 React Flow。

而整个执行层寄生在已有的 agent 轮询之上、节点调用直连平台 server lib—— 没有新造运行时、没有新的守护进程，却让一张拖出来的图，能原生驱动 RandallFlare 从 Worker 到 Analytics 的每一个原语。