绝尘
4aa689e110
- StateMachine 内部 HashMap→Arc<Mutex<HashMap>>，clone 共享底层 - DagExecutor 加 state_machine() 访问器 - AppState 加 workflow_state_registry（execution_id→StateMachine） - run_workflow 注册执行器状态机 + 完成清理 - cancel_workflow_node IPC 经 execution_id 取共享引用 set_cancelled，直达 HumanNode 轮询 - 前端取消按钮 + cancelHumanApproval store 方法 - 加 clone_shares 单测验证共享语义
224 lines
7.5 KiB
Rust
224 lines
7.5 KiB
Rust
//! 工作流相关命令 — 触发执行、查询执行记录
|
||
//!
|
||
//! 事件转发：run_workflow 在执行前订阅事件总线，
|
||
//! 将 WorkflowEvent 包装为 `{ execution_id, event }` 通过 `workflow-event` 事件发给前端。
|
||
|
||
use serde::Serialize;
|
||
use tauri::{AppHandle, Emitter, State};
|
||
use tokio::sync::broadcast::error::RecvError;
|
||
|
||
use df_core::events::{WorkflowEvent, HumanApprovalResponse};
|
||
use df_core::types::new_id;
|
||
use df_storage::crud::WorkflowRepo;
|
||
use df_storage::models::WorkflowRecord;
|
||
use df_workflow::dag_def::DagDef;
|
||
use df_workflow::executor::DagExecutor;
|
||
|
||
use crate::state::AppState;
|
||
|
||
use super::now_millis;
|
||
|
||
/// 转发到前端的事件载荷
|
||
#[derive(Debug, Clone, Serialize)]
|
||
struct WorkflowEventPayload {
|
||
/// 工作流执行记录 ID
|
||
execution_id: String,
|
||
/// 原始工作流事件（serde tag = "type"）
|
||
event: WorkflowEvent,
|
||
}
|
||
|
||
/// 触发工作流执行（核心命令）
|
||
///
|
||
/// 流程：build_dag 校验 → 写入执行记录(status=running) → 后台异步执行 →
|
||
/// 事件经 EventBus 转发到前端 → 完成后更新执行记录状态。
|
||
/// 立即返回执行记录 ID，前端凭此关联后续事件。
|
||
#[tauri::command]
|
||
pub async fn run_workflow(
|
||
app: AppHandle,
|
||
state: State<'_, AppState>,
|
||
name: String,
|
||
dag: DagDef,
|
||
config: serde_json::Value,
|
||
) -> Result<String, String> {
|
||
// 1. 先构建运行时 DAG，校验失败直接返回，不落库
|
||
let runtime_dag = state.registry.build_dag(&dag).map_err(|e| e.to_string())?;
|
||
|
||
// 2. 写入执行记录（status=running）
|
||
let execution_id = new_id();
|
||
let dag_json = serde_json::to_string(&dag).map_err(|e| e.to_string())?;
|
||
let record = WorkflowRecord {
|
||
id: execution_id.clone(),
|
||
name,
|
||
dag_json,
|
||
status: "running".to_string(),
|
||
triggered_by: Some("manual".to_string()),
|
||
project_id: None,
|
||
task_id: None,
|
||
created_at: now_millis(),
|
||
completed_at: None,
|
||
};
|
||
state
|
||
.workflows
|
||
.insert(record)
|
||
.await
|
||
.map_err(|e| e.to_string())?;
|
||
|
||
// 3. 订阅事件总线，把事件转发给前端（收到完成/失败事件后退出）
|
||
let mut rx = state.event_bus.subscribe();
|
||
let forward_app = app.clone();
|
||
let forward_exec_id = execution_id.clone();
|
||
tauri::async_runtime::spawn(async move {
|
||
loop {
|
||
match rx.recv().await {
|
||
Ok(event) => {
|
||
let finished = matches!(
|
||
event,
|
||
WorkflowEvent::WorkflowCompleted { .. }
|
||
| WorkflowEvent::WorkflowFailed { .. }
|
||
);
|
||
let payload = WorkflowEventPayload {
|
||
execution_id: forward_exec_id.clone(),
|
||
event,
|
||
};
|
||
if let Err(e) = forward_app.emit("workflow-event", &payload) {
|
||
tracing::warn!("工作流事件转发失败: {}", e);
|
||
}
|
||
if finished {
|
||
break;
|
||
}
|
||
}
|
||
// 消费过慢丢失部分事件时继续接收
|
||
Err(RecvError::Lagged(n)) => {
|
||
tracing::warn!("工作流事件消费滞后，丢失 {} 条", n);
|
||
}
|
||
Err(RecvError::Closed) => break,
|
||
}
|
||
}
|
||
});
|
||
|
||
// 4. 后台异步执行 DAG，完成后更新执行记录状态
|
||
let event_bus = state.event_bus.clone();
|
||
let db = state.db.clone();
|
||
let exec_id = execution_id.clone();
|
||
let state_registry = state.workflow_state_registry.clone();
|
||
tauri::async_runtime::spawn(async move {
|
||
let mut executor = DagExecutor::new(event_bus.clone(), exec_id.clone());
|
||
// 注册执行器状态机：StateMachine 内部 Arc<Mutex>，clone 共享底层 HashMap，
|
||
// cancel_workflow_node IPC 经 execution_id 取此引用 set_cancelled，直达运行中 HumanNode
|
||
state_registry
|
||
.lock()
|
||
.await
|
||
.insert(exec_id.clone(), executor.state_machine());
|
||
let result = executor.run(&runtime_dag, config).await;
|
||
|
||
let (status, error) = match &result {
|
||
Ok(_) => ("completed", None),
|
||
Err(e) => ("failed", Some(format!("{:#}", e))),
|
||
};
|
||
|
||
// 更新执行记录（Repo 内部仅持有 Arc 连接，重建开销可忽略）
|
||
let workflows = WorkflowRepo::new(&db);
|
||
if let Err(e) = workflows.update_field(&exec_id, "status", status).await {
|
||
tracing::error!("更新工作流状态失败: {}", e);
|
||
}
|
||
if let Err(e) = workflows
|
||
.update_field(&exec_id, "completed_at", &now_millis())
|
||
.await
|
||
{
|
||
tracing::error!("更新工作流完成时间失败: {}", e);
|
||
}
|
||
|
||
// 执行结束（成功/失败）清理状态注册表，防内存泄漏
|
||
state_registry.lock().await.remove(&exec_id);
|
||
|
||
// 执行失败时补发 WorkflowFailed（执行器内部只发 NodeFailed）
|
||
if let Some(error) = error {
|
||
event_bus
|
||
.send(WorkflowEvent::WorkflowFailed {
|
||
error,
|
||
failed_node: String::new(),
|
||
})
|
||
.await;
|
||
}
|
||
});
|
||
|
||
Ok(execution_id)
|
||
}
|
||
|
||
/// 列出全部工作流执行记录
|
||
#[tauri::command]
|
||
pub async fn list_workflow_executions(
|
||
state: State<'_, AppState>,
|
||
) -> Result<Vec<WorkflowRecord>, String> {
|
||
state.workflows.list_all().await.map_err(|e| e.to_string())
|
||
}
|
||
|
||
/// 按 ID 查询工作流执行记录
|
||
#[tauri::command]
|
||
pub async fn get_workflow_execution(
|
||
state: State<'_, AppState>,
|
||
id: String,
|
||
) -> Result<Option<WorkflowRecord>, String> {
|
||
state
|
||
.workflows
|
||
.get_by_id(&id)
|
||
.await
|
||
.map_err(|e| e.to_string())
|
||
}
|
||
|
||
/// 发送人工审批响应
|
||
#[tauri::command]
|
||
pub async fn approve_human_approval(
|
||
app: AppHandle,
|
||
state: State<'_, AppState>,
|
||
execution_id: String,
|
||
node_id: String,
|
||
decision: String,
|
||
comment: Option<String>,
|
||
) -> Result<(), String> {
|
||
// 发送审批响应到事件总线
|
||
let response = HumanApprovalResponse {
|
||
execution_id: execution_id.clone(),
|
||
node_id: node_id.clone(),
|
||
decision,
|
||
comment,
|
||
};
|
||
|
||
let event = WorkflowEvent::HumanApprovalResponse {
|
||
execution_id: response.execution_id,
|
||
node_id: response.node_id,
|
||
decision: response.decision,
|
||
comment: response.comment,
|
||
};
|
||
|
||
// 直接发送到全局事件总线
|
||
state.event_bus.send(event).await;
|
||
Ok(())
|
||
}
|
||
|
||
/// 取消工作流节点（人工审批取消）
|
||
///
|
||
/// 从执行器状态机注册表取出共享引用，调 `set_cancelled` 置目标节点为 Cancelled。
|
||
/// StateMachine 内部 Arc<Mutex> 共享底层 HashMap，IPC 写入直达运行中阻塞节点（HumanNode）
|
||
/// 的 select! 轮询分支，其 is_cancelled 检测到后返回 Err "人工审批被取消"。
|
||
#[tauri::command]
|
||
pub async fn cancel_workflow_node(
|
||
state: State<'_, AppState>,
|
||
execution_id: String,
|
||
node_id: String,
|
||
) -> Result<(), String> {
|
||
// 经 execution_id 取执行器状态机（与 NodeContext.node_status 共享同一 HashMap）
|
||
let registry = state.workflow_state_registry.lock().await;
|
||
let sm = match registry.get(&execution_id) {
|
||
Some(sm) => sm,
|
||
None => return Err(format!("工作流 {} 不存在或已结束", execution_id)),
|
||
};
|
||
sm.set_cancelled(node_id.clone());
|
||
tracing::info!(
|
||
"工作流节点取消：execution_id={}, node_id={}",
|
||
execution_id,
|
||
node_id
|
||
);
|
||
Ok(())
|
||
}
|