lxy/DevFlow
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

重构: 巨函数拆分 + 清理历史标记注释 + custom_prompt/停止按钮/tunnel 改进

Browse Source
This commit is contained in:
lxy
2026-07-31 21:36:12 +08:00
parent 365af554da
commit bd9031d35d
73 changed files with 1421 additions and 1064 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src-tauri/src/commands/workflow.rs
+65 -52
View File
@@ -34,6 +34,53 @@ struct WorkflowEventPayload {
event: WorkflowEvent,
}
/// Lagged 兜底：查 DB 终态，命中返回 `(要补发的事件, status)`；未命中返回 None。
///
/// broadcast `Lagged` 不暴露被丢事件类型，关键终态事件（WorkflowCompleted/Failed）可能已丢，
/// forward 循环会永久等不到。达阈值时查 DB 终态补发。三种"未命中"（仍 running / 无记录 /
/// 查询失败）统一返回 None 并各自 warn，主循环据此重置 `lagged_total` 继续等。
async fn probe_lagged_terminal(
db: &df_storage::db::Database,
exec_id: &ExecutionId,
) -> Option<(WorkflowEvent, String)> {
let record = match WorkflowRepo::new(db).get_by_id(exec_id).await {
Ok(Some(r)) => r,
Ok(None) => {
tracing::warn!(
execution_id = %exec_id,
"Lagged 兜底查询未找到执行记录，继续等待事件"
);
return None;
}
Err(e) => {
tracing::warn!(
execution_id = %exec_id,
error = %e,
"Lagged 兜底查询 DB 失败，继续等待事件"
);
return None;
}
};
let event = match record.status.as_str() {
"completed" => WorkflowEvent::WorkflowCompleted {
execution_id: exec_id.clone(),
total_duration_ms: 0,
},
"failed" => WorkflowEvent::WorkflowFailed {
execution_id: exec_id.clone(),
error: "工作流执行失败（Lagged 兜底补发，详情见 DB）".to_string(),
failed_node: String::new(),
},
"cancelled" => WorkflowEvent::WorkflowFailed {
execution_id: exec_id.clone(),
error: "工作流被取消（Lagged 兜底补发）".to_string(),
failed_node: String::new(),
},
_ => return None,
};
Some((event, record.status))
}
/// 触发工作流执行（核心命令）
///
/// 流程：build_dag 校验 → 写入执行记录(status=running) → 后台异步执行 →
@@ -183,63 +230,29 @@ pub async fn run_workflow_inner(
if lagged_total < LAGGED_PROBE_THRESHOLD {
continue;
}
// 累计/单次达阈值：查 DB 终态兜底
let workflows = WorkflowRepo::new(&forward_db);
match workflows.get_by_id(&forward_exec_id).await {
Ok(Some(record)) => {
let terminal = match record.status.as_str() {
"completed" => Some(WorkflowEvent::WorkflowCompleted {
execution_id: forward_exec_id.clone(),
total_duration_ms: 0,
}),
"failed" => Some(WorkflowEvent::WorkflowFailed {
execution_id: forward_exec_id.clone(),
error: "工作流执行失败（Lagged 兜底补发，详情见 DB）"
.to_string(),
failed_node: String::new(),
}),
"cancelled" => Some(WorkflowEvent::WorkflowFailed {
execution_id: forward_exec_id.clone(),
error: "工作流被取消（Lagged 兜底补发）".to_string(),
failed_node: String::new(),
}),
_ => None,
// 累计/单次达阈值：查 DB 终态兜底（probe_lagged_terminal 收口三种未命中情况）
match probe_lagged_terminal(&forward_db, &forward_exec_id).await {
Some((synth_event, status)) => {
let payload = WorkflowEventPayload {
execution_id: forward_exec_id.clone(),
event: synth_event.clone(),
};
if let Some(synth_event) = terminal {
let payload = WorkflowEventPayload {
execution_id: forward_exec_id.clone(),
event: synth_event.clone(),
};
if let Err(e) = forward_app.emit("workflow-event", &payload) {
tracing::warn!("Lagged 终态兜底事件转发失败: {}", e);
}
tracing::info!(
execution_id = %forward_exec_id,
status = %record.status,
"Lagged 兜底命中终态，补发 {} 并退出 forward",
match synth_event {
WorkflowEvent::WorkflowCompleted { .. } => "WorkflowCompleted",
_ => "WorkflowFailed",
}
);
break;
if let Err(e) = forward_app.emit("workflow-event", &payload) {
tracing::warn!("Lagged 终态兜底事件转发失败: {}", e);
}
// DB 仍非终态（running），重置累计计数继续等待后续事件
lagged_total = 0;
}
Ok(None) => {
tracing::warn!(
tracing::info!(
execution_id = %forward_exec_id,
"Lagged 兜底查询未找到执行记录，继续等待事件"
status = %status,
"Lagged 兜底命中终态，补发 {} 并退出 forward",
match synth_event {
WorkflowEvent::WorkflowCompleted { .. } => "WorkflowCompleted",
_ => "WorkflowFailed",
}
);
lagged_total = 0;
break;
}
Err(e) => {
tracing::warn!(
execution_id = %forward_exec_id,
error = %e,
"Lagged 兜底查询 DB 失败，继续等待事件"
);
None => {
// 未命中(仍 running / 无记录 / 查询失败):重置累计计数继续等待
lagged_total = 0;
}
}