新增: 工作流审批取消端到端(StateMachine Arc 共享 + 取消 IPC 注册表)

- 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 单测验证共享语义
This commit is contained in:
2026-06-14 15:19:36 +08:00
parent 0789a36030
commit 4aa689e110
7 changed files with 166 additions and 19 deletions

View File

@@ -33,6 +33,15 @@ impl DagExecutor {
}
}
/// 返回状态机的共享引用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 工作流
///
/// 同一拓扑层内的节点并发执行,层与层之间串行;

View File

@@ -2,29 +2,38 @@
//!
//! 合法转换Pending → RunningRunning → Completed / Failed
//! 非法转换返回错误,避免状态被随意覆盖。
//!
//! 共享语义:内部用 `Arc<Mutex<HashMap>>``clone()` 为浅拷贝Arc 引用计数 +1
//! 所有 clone 共享同一底层 HashMap。DagExecutor 的 `state_machine` 与其下沉到
//! `NodeContext.node_status` 的 clone 共享底层run_workflow 把执行器状态机注册到
//! AppState 全局表后,`cancel_workflow_node` IPC 的 `set_cancelled` 可直达运行中
//! 阻塞节点(如 HumanNode的 `is_cancelled` 轮询。
use std::collections::HashMap;
use std::sync::{Arc, Mutex};
use df_core::types::{NodeId, NodeStatus};
/// 节点状态机
#[derive(Debug, Clone)]
pub struct StateMachine {
/// 节点状态映射
states: HashMap<NodeId, NodeStatus>,
/// 节点状态映射Arc 共享clone 浅拷贝同一 HashMap
states: Arc<Mutex<HashMap<NodeId, NodeStatus>>>,
}
impl StateMachine {
/// 创建状态机
pub fn new() -> Self {
Self {
states: HashMap::new(),
states: Arc::new(Mutex::new(HashMap::new())),
}
}
/// 获取节点状态
pub fn get(&self, node_id: &NodeId) -> NodeStatus {
self.states
.lock()
.expect("状态机锁中毒")
.get(node_id)
.cloned()
.unwrap_or(NodeStatus::Pending)
@@ -41,8 +50,12 @@ impl StateMachine {
}
/// 状态转换 — 校验合法性后更新,非法转换返回错误
pub fn transition(&mut self, node_id: NodeId, target: NodeStatus) -> anyhow::Result<()> {
let current = self.get(&node_id);
pub fn transition(&self, node_id: NodeId, target: NodeStatus) -> anyhow::Result<()> {
let mut states = self.states.lock().expect("状态机锁中毒");
let current = states
.get(&node_id)
.cloned()
.unwrap_or(NodeStatus::Pending);
if !Self::is_legal(&current, &target) {
anyhow::bail!(
"节点 {} 状态转换非法:{} -> {}",
@@ -51,38 +64,55 @@ impl StateMachine {
target.as_str()
);
}
self.states.insert(node_id, target);
states.insert(node_id, target);
Ok(())
}
/// 设置节点为运行中(仅允许 Pending -> Running
pub fn set_running(&mut self, node_id: NodeId) -> anyhow::Result<()> {
pub fn set_running(&self, node_id: NodeId) -> anyhow::Result<()> {
self.transition(node_id, NodeStatus::Running)
}
/// 设置节点为已完成(仅允许 Running -> Completed
pub fn set_completed(&mut self, node_id: NodeId) -> anyhow::Result<()> {
pub fn set_completed(&self, node_id: NodeId) -> anyhow::Result<()> {
self.transition(node_id, NodeStatus::Completed)
}
/// 设置节点为失败(仅允许 Running -> Failed
pub fn set_failed(&mut self, node_id: NodeId) -> anyhow::Result<()> {
pub fn set_failed(&self, node_id: NodeId) -> anyhow::Result<()> {
self.transition(node_id, NodeStatus::Failed)
}
/// 设置节点为等待中(暂未纳入转换校验)
pub fn set_waiting(&mut self, node_id: NodeId) {
self.states.insert(node_id, NodeStatus::Waiting);
pub fn set_waiting(&self, node_id: NodeId) {
self.states
.lock()
.expect("状态机锁中毒")
.insert(node_id, NodeStatus::Waiting);
}
/// 设置节点为已跳过(暂未纳入转换校验)
pub fn set_skipped(&mut self, node_id: NodeId) {
self.states.insert(node_id, NodeStatus::Skipped);
pub fn set_skipped(&self, node_id: NodeId) {
self.states
.lock()
.expect("状态机锁中毒")
.insert(node_id, NodeStatus::Skipped);
}
/// 获取所有状态快照
pub fn snapshot(&self) -> &HashMap<NodeId, NodeStatus> {
&self.states
/// 设置节点为已取消(暂未纳入转换校验,同 set_waiting/set_skipped 模式)
///
/// 人工审批取消场景由前端 cancel_workflow_node IPC 触发;
/// 阻塞节点(如 HumanNode)在 select! 轮询分支通过 is_cancelled 检测到后返回 Err。
pub fn set_cancelled(&self, node_id: NodeId) {
self.states
.lock()
.expect("状态机锁中毒")
.insert(node_id, NodeStatus::Cancelled);
}
/// 获取所有状态快照clone 返回,调用方持独立副本)
pub fn snapshot(&self) -> HashMap<NodeId, NodeStatus> {
self.states.lock().expect("状态机锁中毒").clone()
}
/// 检查节点是否被取消
@@ -103,7 +133,7 @@ mod tests {
#[test]
fn test_legal_transitions() {
let mut sm = StateMachine::new();
let sm = StateMachine::new();
let id = "node-a".to_string();
// Pending -> Running -> Completed 全程合法
@@ -121,7 +151,7 @@ mod tests {
#[test]
fn test_illegal_transitions_rejected() {
let mut sm = StateMachine::new();
let sm = StateMachine::new();
let id = "node-a".to_string();
// Pending -> Completed 非法
@@ -144,4 +174,41 @@ mod tests {
sm.set_running(id2.clone()).unwrap();
assert!(sm.set_running(id2.clone()).is_err());
}
#[test]
fn test_set_cancelled_marks_node_cancelled() {
// set_cancelled 直接置位,不经转换校验(同 set_waiting/set_skipped 模式)
let sm = StateMachine::new();
let id = "human-1".to_string();
// 初始为 Pendingis_cancelled = false
assert_eq!(sm.get(&id), NodeStatus::Pending);
assert!(!sm.is_cancelled(&id));
// 取消后置位 Cancelledis_cancelled = true
sm.set_cancelled(id.clone());
assert_eq!(sm.get(&id), NodeStatus::Cancelled);
assert!(sm.is_cancelled(&id));
}
#[test]
fn test_clone_shares_underlying_state() {
// 核心机制clone 浅拷贝 Arc写入对另一副本可见。
// 这让 cancel IPC 写 AppState 注册表的 StateMachine
// 直达 HumanNode 持有的 ctx.node_statusexecutor.state_machine 的 clone
let sm = StateMachine::new();
let sm_clone = sm.clone();
// 原实例写入clone 可见
sm.set_cancelled("human-1".to_string());
assert!(sm_clone.is_cancelled(&"human-1".to_string()));
// clone 写入,原实例亦可见(双向)
sm_clone.set_running("node-a".to_string()).unwrap();
assert_eq!(sm.get(&"node-a".to_string()), NodeStatus::Running);
// 第三副本同样共享
let sm3 = sm_clone.clone();
assert!(sm3.is_cancelled(&"human-1".to_string()));
}
}