修复: B-03b-R1 取消节点跳 set_failed + R2 取消测试(补完 B-03b 复核)

R1: executor Err 处理加 is_cancelled 检测,已取消节点跳 set_failed(Cancelled→Failed transition 非法会 bail 致工作流崩溃),状态保 Cancelled,emit NodeFailed 仍发
R2: test_cancelled_node_skips_set_failed 验证取消传播链(CancelSelfNode 共享 node_status 自取消→Err→executor 不 bail→状态 Cancelled)
补完 B-03b 端到端:4aa689e 只通 IPC→set_cancelled→HumanNode→Err 链,漏验 HumanNode→executor Err 处理与 Cancelled 冲突;df-workflow 14 test pass
This commit is contained in:
2026-06-14 16:10:10 +08:00
parent 406b9216de
commit d6cd2805be
2 changed files with 58 additions and 4 deletions

View File

@@ -121,11 +121,16 @@ impl DagExecutor {
outputs.insert(node_id, output);
}
Err(e) => {
self.state_machine.set_failed(node_id.clone())?;
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: e.to_string(),
error: error_msg,
})
.await;
// 同层多个失败时只报告第一个
@@ -248,4 +253,53 @@ mod tests {
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
);
}
}