259 lines
13 KiB
Rust
259 lines
13 KiB
Rust
//! DAG 执行器 — 按拓扑层级调度执行
|
||
|
||
use std::collections::HashMap;
|
||
|
||
use df_types::events::WorkflowEvent;
|
||
use df_types::types::{ExecutionId, NodeId};
|
||
|
||
use crate::conditions::ConditionEngine;
|
||
use crate::dag::Dag;
|
||
use crate::eventbus::EventBus;
|
||
use crate::node::{NodeContext, NodeOutput};
|
||
use crate::state::StateMachine;
|
||
|
||
/// DAG 执行器
|
||
pub struct DagExecutor {
|
||
/// 事件总线
|
||
event_bus: EventBus,
|
||
/// 节点状态机
|
||
state_machine: StateMachine,
|
||
/// 工作流执行 ID（由调用方传入，下沉到每个 NodeContext）
|
||
execution_id: ExecutionId,
|
||
}
|
||
|
||
impl DagExecutor {
|
||
/// 创建执行器
|
||
///
|
||
/// `execution_id` 为本次工作流执行的唯一标识，会下沉到每个节点的 NodeContext，
|
||
/// 用于节点内的事件关联、审计追踪等。
|
||
pub fn new(event_bus: EventBus, execution_id: ExecutionId) -> Self {
|
||
Self {
|
||
event_bus,
|
||
state_machine: StateMachine::new(),
|
||
execution_id,
|
||
}
|
||
}
|
||
|
||
/// 返回状态机的共享引用（Arc clone，与 NodeContext.node_status 共享底层 HashMap）
|
||
///
|
||
/// run_workflow 创建执行器后将其注册到 AppState 全局表，
|
||
/// cancel_workflow_node IPC 经 execution_id 取出此引用调 set_cancelled，
|
||
/// 写入直达运行中阻塞节点（HumanNode）的 is_cancelled 轮询。
|
||
pub fn state_machine(&self) -> StateMachine {
|
||
self.state_machine.clone()
|
||
}
|
||
|
||
/// 执行 DAG 工作流
|
||
///
|
||
/// 同一拓扑层内的节点并发执行，层与层之间串行；
|
||
/// 本层全部节点完成后统一更新状态，任一失败则中止后续层。
|
||
pub async fn run(
|
||
&mut self,
|
||
dag: &Dag,
|
||
initial_config: serde_json::Value,
|
||
) -> anyhow::Result<HashMap<NodeId, NodeOutput>> {
|
||
let layers = dag.topological_layers()?;
|
||
let mut outputs: HashMap<NodeId, NodeOutput> = HashMap::new();
|
||
let start = std::time::Instant::now();
|
||
|
||
tracing::info!("DAG 执行开始，共 {} 层", layers.len());
|
||
|
||
// 预建入边索引：target → 直接前驱列表 + 边条件表达式(O(E) 一次构建)
|
||
// 避免在内层按节点循环中调用 dag.predecessors()（每次 O(E) 全表扫描，整体 O(V·E)）。
|
||
// 含 condition 是为 conditions-eval feature 接入:flag 开时按入边 condition 过滤前驱 input。
|
||
// flag 关时 condition 字段被忽略,行为与增强前完全一致(零破坏)。
|
||
let mut adjacency_in: HashMap<NodeId, Vec<(NodeId, Option<String>)>> = HashMap::new();
|
||
for edge in &dag.edges {
|
||
adjacency_in
|
||
.entry(edge.target.clone())
|
||
.or_default()
|
||
.push((edge.source.clone(), edge.condition.clone()));
|
||
}
|
||
|
||
for (layer_idx, layer) in layers.iter().enumerate() {
|
||
tracing::info!("执行第 {} 层，共 {} 个节点", layer_idx, layer.len());
|
||
|
||
// 阶段一：逐节点发 NodeStarted、置为运行中，并构建执行 future
|
||
let mut node_futures = Vec::with_capacity(layer.len());
|
||
for node_id in layer {
|
||
let node = dag.nodes.get(node_id).ok_or_else(|| {
|
||
anyhow::anyhow!("节点 {} 不存在于 DAG 中", node_id)
|
||
})?;
|
||
|
||
// 构建节点上下文：通过入边索引 O(入度) 取前驱，而非 O(E) 全表扫描。
|
||
//
|
||
// conditions-eval feature(默认关):
|
||
// 关:所有入边 inputs 全收集,condition 字段完全忽略 = 旧行为(零破坏)。
|
||
// 开:对带 condition 的入边以 source output.data 为 context 求值;
|
||
// false 则不收集该前驱 input。若节点【存在带 condition 的入边,且无任何
|
||
// 入边(条件或无条件)放行】,则跳过该节点执行(不入 outputs、保持 Pending、
|
||
// 不 emit),实现条件路由。无条件边 source 有 output 即视为放行。
|
||
// 求值失败保守 false(对齐引擎语义)。
|
||
//
|
||
// 注:跳过判定在 set_running 之前 —— 跳过的节点不应被标记 Running/触发事件,
|
||
// 保持 Pending 终态(与"条件未命中"语义一致)。
|
||
let eval_conditions = cfg!(feature = "conditions-eval");
|
||
let mut inputs: HashMap<NodeId, NodeOutput> = HashMap::new();
|
||
let mut has_any_cond_edge = false;
|
||
let mut any_edge_passed = false; // 任一入边放行(含无条件边)
|
||
if let Some(preds) = adjacency_in.get(node_id) {
|
||
for (pred_id, cond) in preds {
|
||
let Some(out) = outputs.get(pred_id) else {
|
||
continue;
|
||
};
|
||
|
||
// flag 关:condition 完全忽略,旧行为全收集;
|
||
// flag 开 + 无 condition 入边:无条件放行(必收集);
|
||
// flag 开 + 有 condition 入边:以 source output.data 为 context 求值,
|
||
// 失败保守 false(对齐引擎语义)。
|
||
let passed = match (eval_conditions, cond) {
|
||
(false, _) => true,
|
||
(true, None) => true,
|
||
(true, Some(c)) => {
|
||
ConditionEngine::evaluate(c, &out.data).unwrap_or(false)
|
||
}
|
||
};
|
||
if eval_conditions && cond.is_some() {
|
||
has_any_cond_edge = true;
|
||
}
|
||
if passed {
|
||
inputs.insert(pred_id.clone(), out.clone());
|
||
any_edge_passed = true;
|
||
}
|
||
}
|
||
} else {
|
||
// 根节点(无入边):不参与条件跳过判定,正常执行
|
||
any_edge_passed = true;
|
||
}
|
||
|
||
// 条件路由:flag 开 + 存在条件边 + 没有任何入边放行 → 跳过执行
|
||
if eval_conditions && has_any_cond_edge && !any_edge_passed {
|
||
tracing::info!(
|
||
"节点 {} 所有入边条件均不满足,跳过执行(条件路由)",
|
||
node_id
|
||
);
|
||
continue;
|
||
}
|
||
|
||
// 发送 NodeStarted 事件
|
||
self.event_bus
|
||
.send(WorkflowEvent::NodeStarted {
|
||
node_id: node_id.clone(),
|
||
})
|
||
.await;
|
||
|
||
self.state_machine.set_running(node_id.clone())?;
|
||
|
||
let ctx = NodeContext {
|
||
node_id: node_id.clone(),
|
||
inputs,
|
||
// 节点级覆盖全局级：deep_merge(initial_config, node_configs[id])。
|
||
// 节点 NodeDef.config 未定义（node_configs 缺该 key）→ 直接用全局 initial_config，
|
||
// 与旧行为一致（零行为破坏现有 HumanNode/AiNode 等读全局 config 的调用方）。
|
||
config: match dag.node_configs.get(node_id) {
|
||
Some(node_cfg) => crate::dag::deep_merge(&initial_config, node_cfg),
|
||
None => initial_config.clone(),
|
||
},
|
||
execution_id: self.execution_id.clone(),
|
||
event_bus: self.event_bus.clone(),
|
||
// 共享执行器状态机，使节点（如 HumanNode）能读取真实状态而非空状态机
|
||
node_status: self.state_machine.clone(),
|
||
};
|
||
|
||
// 仅捕获节点共享引用与所有权上下文，避免与 self 借用冲突
|
||
let id = node_id.clone();
|
||
node_futures.push(async move {
|
||
let node_start = std::time::Instant::now();
|
||
let result = node.execute(ctx).await;
|
||
(id, result, node_start.elapsed().as_millis() as u64)
|
||
});
|
||
}
|
||
|
||
// 阶段二：同层节点并发执行，等待全部完成
|
||
let results = futures::future::join_all(node_futures).await;
|
||
|
||
// 阶段三：统一更新状态并发送事件，任一失败则整体返回 Err
|
||
let mut first_err: Option<anyhow::Error> = None;
|
||
for (node_id, result, duration) in results {
|
||
match result {
|
||
Ok(output) => {
|
||
// 已取消的节点跳过 set_completed:
|
||
// Ok 后 join_all 让出执行权,窗口内 cancel_workflow_node IPC → set_cancelled 改 Cancelled,
|
||
// 随后此处 set_completed 走 transition 命中(Cancelled→Completed)非法 → bail 致已批准审批报失败(R-P1-3 TOCTOU)
|
||
// 与 Err 分支 is_cancelled 短路对称:已取消节点保持 Cancelled 终态
|
||
if !self.state_machine.is_cancelled(&node_id) {
|
||
self.state_machine.set_completed(node_id.clone())?;
|
||
// 未取消才 emit NodeCompleted,取消时 emit NodeCancelled(对齐 Err 分支语义):
|
||
// 避免 状态机=Cancelled 但事件=NodeCompleted 的矛盾信号
|
||
self.event_bus
|
||
.send(WorkflowEvent::NodeCompleted {
|
||
node_id: node_id.clone(),
|
||
duration_ms: duration,
|
||
})
|
||
.await;
|
||
} else {
|
||
// 对齐 Err 分支:取消节点 emit NodeCancelled(非 NodeCompleted),
|
||
// 前端按 type 归「取消」,与状态机 Cancelled 一致
|
||
self.event_bus
|
||
.send(WorkflowEvent::NodeCancelled {
|
||
node_id: node_id.clone(),
|
||
})
|
||
.await;
|
||
}
|
||
// outputs.insert 保留现状不动:取消时仍写入 output 供下游消费,
|
||
// 当前测试锁定"取消不中止后续层",改 insert 会变行为致回归
|
||
outputs.insert(node_id, output);
|
||
}
|
||
Err(e) => {
|
||
let error_msg = e.to_string();
|
||
// 已取消的节点(如 HumanNode 审批取消)跳过 set_failed:
|
||
// Cancelled 已是终态,transition(Cancelled→Failed) 非法会 bail 致工作流崩溃(b-R1)
|
||
if self.state_machine.is_cancelled(&node_id) {
|
||
// emit NodeCancelled(非 NodeFailed):取消语义有别于失败,前端按 type 归「取消」非「失败」
|
||
self.event_bus
|
||
.send(WorkflowEvent::NodeCancelled {
|
||
node_id: node_id.clone(),
|
||
})
|
||
.await;
|
||
} else {
|
||
self.state_machine.set_failed(node_id.clone())?;
|
||
self.event_bus
|
||
.send(WorkflowEvent::NodeFailed {
|
||
node_id: node_id.clone(),
|
||
error: error_msg,
|
||
})
|
||
.await;
|
||
}
|
||
// 取消与失败一致:节点返回 Err 即中止后续层(run 返回 Err);
|
||
// 仅事件类型区分,工作流结果语义保持不变(零行为破坏)。
|
||
if first_err.is_none() {
|
||
first_err = Some(e.context(format!("节点 {} 执行失败", node_id)));
|
||
}
|
||
}
|
||
}
|
||
}
|
||
|
||
if let Some(e) = first_err {
|
||
return Err(e);
|
||
}
|
||
}
|
||
|
||
let total = start.elapsed().as_millis() as u64;
|
||
// b-R10 ③: WorkflowCompleted 携带 execution_id,供 forward 循环按 exec_id 匹配
|
||
// (全局 event_bus 单例下并发工作流的终态事件不再误触发他人 forward 的完成判定)。
|
||
self.event_bus
|
||
.send(WorkflowEvent::WorkflowCompleted {
|
||
execution_id: self.execution_id.clone(),
|
||
total_duration_ms: total,
|
||
})
|
||
.await;
|
||
|
||
tracing::info!("DAG 执行完成，耗时 {}ms", total);
|
||
Ok(outputs)
|
||
}
|
||
}
|
||
|
||
#[cfg(test)]
|
||
#[path = "executor_helpers.rs"]
|
||
mod tests;
|