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)$ ：

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

4.2 边状态与就绪判定#

引擎不做全局拓扑排序，而是从已完成的 step 推导每条边的状态。设为节点的执行状态（未完成 / OK / FAILED / SKIPPED），并对分支类节点记其「取用句柄」。则一条边的状态：

\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"，于是边 dead、边 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：

4.3 波次遍历算法#

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

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

4.4 节点输入的合并#

一个就绪节点的输入，取它的源节点输出。设，则：

\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)$ 为从节点集沿边前向可达的节点集，则循环体：

\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 的核心（整字段单 {{ }} 保留原始类型，否则字符串插值）：

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

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

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

崩溃恢复：租约过期（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)

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

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 略）：

执行轨迹：

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

九、小结#

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

后端：不做全局拓扑排序，而是用边状态 + 就绪判定做增量的波次遍历——天然支持分支剪枝、循环内联、跨节点上下文，且每一步都是可从数据库无损重建的幂等操作，配合条件更新的乐观租约，得到崩溃自愈、不重复、可同步可异步的持久化执行。
前端：把「整理乱图」归约成经典的分层绘图，用 dagre 的网络单纯形分层 + 中位数排序 + 坐标最优化求解，再做一步中心→左上角的坐标变换接进 React Flow。

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

执行跑在控制面（plane），不在 agent	全 TypeScript，迭代快；节点调用就是对原语的 server-side 直调，owner 隔离天然成立
用 `after()` 挂在已有的 agent 轮询上驱动	零 agent 改动，常驻 Node 进程可靠执行，租约保证不重复推进
图存 JSON，运行时解释	Workflow 引擎是「按 worker 编译」的，装不下动态图；解释器方案把 DAG 逻辑留在 TS 里
节点调用 = server lib 直调	和 v1 REST API 同一套底层函数，不走 HTTP，无 token 周转

`{{ $json.field }}`	本节点输入
`{{ $node['查用户'].email }}`	任意先前节点的输出（按 label / id）—— 跨节点
`{{ $trigger.userId }}`	触发输入
`{{ $item }}`	循环当前项
`{{ $now }}`	ISO 时间戳

Trigger	方块	manual / cron / webhook	—
Worker	方块	HTTP 调 worker 公网 hostname	`fetch`
HTTP	方块	任意 HTTP 请求	`fetch`
Branch	方块（双出口）	条件路由 `true`/`false`	表达式求值
Loop	方块（`each`/`done`）	集合/计数循环，内联子图	§4.6
Transform	方块	JSONata 数据整形	`jsonata`
Subflow	方块	内联另一张 Flow	§4.7

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

一、Flow 是什么#

二、整体架构#

三、数据模型#

四、后端：DAG 执行引擎#

4.1 图的表示#

4.2 边状态与就绪判定#

4.3 波次遍历算法#

4.4 节点输入的合并#

4.5 分支（branch）#

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

4.7 子流程（subflow）#

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

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

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

5.1 问题#

5.2 Sugiyama 分层绘图#

5.3 dagre 集成 + 坐标变换#

5.4 实时运行叠加#

六、数据流全景#

七、节点类型总览#

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

九、小结#

no comments yet

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

一、Flow 是什么#

二、整体架构#

三、数据模型#

四、后端：DAG 执行引擎#

4.1 图的表示#

4.2 边状态与就绪判定#

4.3 波次遍历算法#

4.4 节点输入的合并#

4.5 分支（branch）#

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

4.7 子流程（subflow）#

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

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

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

5.1 问题#

5.2 Sugiyama 分层绘图#

5.3 dagre 集成 + 坐标变换#

5.4 实时运行叠加#

六、数据流全景#

七、节点类型总览#

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

九、小结#

no comments yet