绝尘
2de0c6ecb7
- df-ai: context 历史中毒三档自愈 sanitize_messages(AC3)+anthropic_compat tool_use_id None 跳过(AC1/AC2)+删 router/stream 死码
- df-core: events 加 select_type+decisions 多选审批契约(F-260615-01)
- df-execute: shell run_command 工具复用(F-260615-05)
- df-nodes: human_node 多选校验+2 端到端测(F-01)+取消跳 set_failed(B-03b-R1/R2/R8)
- df-workflow: executor/dag/state cancel 闭环(B-06/07/03a/b)+provider approve options(R-PD-5)
- df-storage: find_path_conflict 抽公共(R-PD-11)+COLS 常量断言
- df-ideas: 删 IdeaPromoter/PromotionPolicy 死码(R-PD-14)
- src-tauri/commands/ai: secret keyring 迁移(FR-S1/R-PD-4)+GeneratingGuard RAII+disarm(B-09/26)+newConversation 软复位(B-10)+stream 心跳/stop select/空回复判错(B-02/04/05/15)+run_command(F-05)+mask audit(AR-3)
- src-tauri/commands/{project,task,workflow,mod,lib,state}: task detail IPC(F-02)+approve decisions+task list 联动(B-29)
- Cargo.lock+Cargo.toml 依赖同步
373 lines
14 KiB
Rust
373 lines
14 KiB
Rust
//! DAG 执行器 — 按拓扑层级调度执行
|
||
|
||
use std::collections::HashMap;
|
||
|
||
use df_core::events::WorkflowEvent;
|
||
use df_core::types::NodeId;
|
||
|
||
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: String,
|
||
}
|
||
|
||
impl DagExecutor {
|
||
/// 创建执行器
|
||
///
|
||
/// `execution_id` 为本次工作流执行的唯一标识，会下沉到每个节点的 NodeContext，
|
||
/// 用于节点内的事件关联、审计追踪等。
|
||
pub fn new(event_bus: EventBus, execution_id: String) -> 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)）。
|
||
let mut adjacency_in: HashMap<NodeId, Vec<NodeId>> = HashMap::new();
|
||
for edge in &dag.edges {
|
||
adjacency_in
|
||
.entry(edge.target.clone())
|
||
.or_default()
|
||
.push(edge.source.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)
|
||
})?;
|
||
|
||
// 发送 NodeStarted 事件
|
||
self.event_bus
|
||
.send(WorkflowEvent::NodeStarted {
|
||
node_id: node_id.clone(),
|
||
})
|
||
.await;
|
||
|
||
self.state_machine.set_running(node_id.clone())?;
|
||
|
||
// 构建节点上下文：通过入边索引 O(入度) 取前驱，而非 O(E) 全表扫描
|
||
let mut inputs = HashMap::new();
|
||
if let Some(preds) = adjacency_in.get(node_id) {
|
||
for pred_id in preds {
|
||
if let Some(out) = outputs.get(pred_id) {
|
||
inputs.insert(pred_id.clone(), out.clone());
|
||
}
|
||
}
|
||
}
|
||
|
||
let ctx = NodeContext {
|
||
node_id: node_id.clone(),
|
||
inputs,
|
||
config: 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())?;
|
||
}
|
||
self.event_bus
|
||
.send(WorkflowEvent::NodeCompleted {
|
||
node_id: node_id.clone(),
|
||
duration_ms: duration,
|
||
})
|
||
.await;
|
||
outputs.insert(node_id, output);
|
||
}
|
||
Err(e) => {
|
||
let error_msg = e.to_string();
|
||
// 已取消的节点(如 HumanNode 审批取消)跳过 set_failed:
|
||
// Cancelled 已是终态,transition(Cancelled→Failed) 非法会 bail 致工作流崩溃(B-03b-R1)
|
||
if !self.state_machine.is_cancelled(&node_id) {
|
||
self.state_machine.set_failed(node_id.clone())?;
|
||
}
|
||
self.event_bus
|
||
.send(WorkflowEvent::NodeFailed {
|
||
node_id: node_id.clone(),
|
||
error: error_msg,
|
||
})
|
||
.await;
|
||
// 同层多个失败时只报告第一个
|
||
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;
|
||
self.event_bus
|
||
.send(WorkflowEvent::WorkflowCompleted {
|
||
total_duration_ms: total,
|
||
})
|
||
.await;
|
||
|
||
tracing::info!("DAG 执行完成，耗时 {}ms", total);
|
||
Ok(outputs)
|
||
}
|
||
}
|
||
|
||
#[cfg(test)]
|
||
mod tests {
|
||
use super::*;
|
||
use crate::node::{Node, NodeResult, NodeSchema};
|
||
use async_trait::async_trait;
|
||
use std::time::Duration;
|
||
|
||
/// 测试节点：sleep 指定毫秒后返回空输出
|
||
struct SleepNode {
|
||
sleep_ms: u64,
|
||
}
|
||
|
||
#[async_trait]
|
||
impl Node for SleepNode {
|
||
async fn execute(&self, _ctx: NodeContext) -> NodeResult {
|
||
tokio::time::sleep(Duration::from_millis(self.sleep_ms)).await;
|
||
Ok(NodeOutput::empty())
|
||
}
|
||
|
||
fn schema(&self) -> NodeSchema {
|
||
NodeSchema {
|
||
params: serde_json::Value::Null,
|
||
output: serde_json::Value::Null,
|
||
}
|
||
}
|
||
|
||
fn node_type(&self) -> &str {
|
||
"sleep"
|
||
}
|
||
}
|
||
|
||
/// 测试节点：直接返回错误
|
||
struct FailNode;
|
||
|
||
#[async_trait]
|
||
impl Node for FailNode {
|
||
async fn execute(&self, _ctx: NodeContext) -> NodeResult {
|
||
Err(anyhow::anyhow!("故意失败"))
|
||
}
|
||
|
||
fn schema(&self) -> NodeSchema {
|
||
NodeSchema {
|
||
params: serde_json::Value::Null,
|
||
output: serde_json::Value::Null,
|
||
}
|
||
}
|
||
|
||
fn node_type(&self) -> &str {
|
||
"fail"
|
||
}
|
||
}
|
||
|
||
#[tokio::test]
|
||
async fn test_same_layer_runs_in_parallel() {
|
||
// 两个无依赖节点位于同一层，各 sleep 100ms
|
||
let mut dag = Dag::new();
|
||
dag.add_node("a".to_string(), Box::new(SleepNode { sleep_ms: 100 }));
|
||
dag.add_node("b".to_string(), Box::new(SleepNode { sleep_ms: 100 }));
|
||
|
||
let mut executor = DagExecutor::new(EventBus::new(), "test-exec".to_string());
|
||
let start = std::time::Instant::now();
|
||
let outputs = executor.run(&dag, serde_json::Value::Null).await.unwrap();
|
||
let elapsed = start.elapsed();
|
||
|
||
assert_eq!(outputs.len(), 2);
|
||
// 串行需要约 200ms，并行应明显小于 180ms
|
||
assert!(
|
||
elapsed < Duration::from_millis(180),
|
||
"同层节点应并行执行，实际耗时 {:?}",
|
||
elapsed
|
||
);
|
||
}
|
||
|
||
#[tokio::test]
|
||
async fn test_layer_failure_aborts_following_layers() {
|
||
// a(失败) 与 b(成功) 同层，c 依赖 a，失败后 c 不应执行
|
||
let mut dag = Dag::new();
|
||
dag.add_node("a".to_string(), Box::new(FailNode));
|
||
dag.add_node("b".to_string(), Box::new(SleepNode { sleep_ms: 10 }));
|
||
dag.add_node("c".to_string(), Box::new(SleepNode { sleep_ms: 10 }));
|
||
dag.add_edge("a".to_string(), "c".to_string());
|
||
|
||
let mut executor = DagExecutor::new(EventBus::new(), "test-exec".to_string());
|
||
let err = executor
|
||
.run(&dag, serde_json::Value::Null)
|
||
.await
|
||
.unwrap_err();
|
||
assert!(err.to_string().contains("节点 a 执行失败"));
|
||
|
||
// 同层成功节点状态正常更新，下游节点保持 Pending
|
||
use df_core::types::NodeStatus;
|
||
assert_eq!(executor.state_machine.get(&"a".to_string()), NodeStatus::Failed);
|
||
assert_eq!(executor.state_machine.get(&"b".to_string()), NodeStatus::Completed);
|
||
assert_eq!(executor.state_machine.get(&"c".to_string()), NodeStatus::Pending);
|
||
}
|
||
|
||
/// 测试节点:执行中通过共享 node_status 自取消(模拟外部 IPC set_cancelled),返回 Err
|
||
struct CancelSelfNode;
|
||
|
||
#[async_trait]
|
||
impl Node for CancelSelfNode {
|
||
async fn execute(&self, ctx: NodeContext) -> NodeResult {
|
||
// 通过共享 node_status 置 Cancelled(写共享 HashMap,executor.state_machine 可见)
|
||
ctx.node_status.set_cancelled(ctx.node_id.clone());
|
||
Err(anyhow::anyhow!("人工审批被取消"))
|
||
}
|
||
|
||
fn schema(&self) -> NodeSchema {
|
||
NodeSchema {
|
||
params: serde_json::Value::Null,
|
||
output: serde_json::Value::Null,
|
||
}
|
||
}
|
||
|
||
fn node_type(&self) -> &str {
|
||
"cancel_self"
|
||
}
|
||
}
|
||
|
||
/// B-03b-R1:取消的节点返回 Err 时,executor 跳过 set_failed(Cancelled→Failed transition 非法会 bail),
|
||
/// 状态保持 Cancelled,run 返回取消相关 Err 而非状态转换错误。
|
||
#[tokio::test]
|
||
async fn test_cancelled_node_skips_set_failed() {
|
||
use df_core::types::NodeStatus;
|
||
let mut dag = Dag::new();
|
||
dag.add_node("x".to_string(), Box::new(CancelSelfNode));
|
||
|
||
let mut executor = DagExecutor::new(EventBus::new(), "test-cancel".to_string());
|
||
let result = executor.run(&dag, serde_json::Value::Null).await;
|
||
|
||
// run 返回 Err(取消致中止后续层),但不 panic/bail transition
|
||
assert!(result.is_err());
|
||
let err = result.unwrap_err().to_string();
|
||
assert!(
|
||
err.contains("取消") || err.contains("执行失败"),
|
||
"应返回取消相关错误,实际: {}",
|
||
err
|
||
);
|
||
// 状态保持 Cancelled(未被 set_failed 覆盖为 Failed)—— R1 修法生效的核心验证
|
||
assert_eq!(
|
||
executor.state_machine.get(&"x".to_string()),
|
||
NodeStatus::Cancelled
|
||
);
|
||
}
|
||
|
||
/// 测试节点:执行中通过共享 node_status 自取消(模拟外部 IPC set_cancelled),返回 Ok
|
||
/// 模拟 HumanNode select! 收合法 HumanApprovalResponse 后 return Ok,但 join_all 让出窗口内
|
||
/// cancel_workflow_node IPC 把节点改 Cancelled 的 TOCTOU 场景。
|
||
struct CancelSelfThenOkNode;
|
||
|
||
#[async_trait]
|
||
impl Node for CancelSelfThenOkNode {
|
||
async fn execute(&self, ctx: NodeContext) -> NodeResult {
|
||
ctx.node_status.set_cancelled(ctx.node_id.clone());
|
||
Ok(NodeOutput::empty())
|
||
}
|
||
|
||
fn schema(&self) -> NodeSchema {
|
||
NodeSchema {
|
||
params: serde_json::Value::Null,
|
||
output: serde_json::Value::Null,
|
||
}
|
||
}
|
||
|
||
fn node_type(&self) -> &str {
|
||
"cancel_self_then_ok"
|
||
}
|
||
}
|
||
|
||
/// R-P1-3:Ok 后取消的 TOCTOU 防护。节点返回 Ok 但执行期间已被 set_cancelled,
|
||
/// executor Ok 分支必须跳过 set_completed(transition Cancelled→Completed 非法会 bail),
|
||
/// 状态保持 Cancelled,run 返回 Ok(已批准审批不应误报失败)。
|
||
#[tokio::test]
|
||
async fn test_cancelled_node_skips_set_completed() {
|
||
use df_core::types::NodeStatus;
|
||
let mut dag = Dag::new();
|
||
dag.add_node("y".to_string(), Box::new(CancelSelfThenOkNode));
|
||
|
||
let mut executor = DagExecutor::new(EventBus::new(), "test-cancel-ok".to_string());
|
||
let result = executor.run(&dag, serde_json::Value::Null).await;
|
||
|
||
// run 返回 Ok —— Ok 节点不应因 TOCTOU 取消被误判为失败
|
||
assert!(
|
||
result.is_ok(),
|
||
"Ok 后取消应短路 set_completed 而非 bail,实际 err: {:?}",
|
||
result.err()
|
||
);
|
||
// 状态保持 Cancelled(未被 set_completed 覆盖为 Completed,也不 bail)—— R-P1-3 修法核心验证
|
||
assert_eq!(
|
||
executor.state_machine.get(&"y".to_string()),
|
||
NodeStatus::Cancelled
|
||
);
|
||
}
|
||
}
|