red-future/ai-agent
4
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(workflow): 错误分类 + in-process 自动重试 + status=1 防重复执行

Browse Source 
Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
This commit is contained in:
19904408334
2026-08-25 16:58:38 +08:00
co-authored by Claude Opus 4.7
parent 95dc0d6052
commit ce2407bb2d
1 changed files with 54 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files
workflow/service/flow/flow_ws_exec.go
+54 -11
View File
@@ -156,6 +156,33 @@ func handleExecute(ctx context.Context, conn *wsCommon.WsConnection, payload int
_ = writeJSON(conn, &wsCommon.WsPushMsg{Type: "ack", Message: fmt.Sprintf("开始执行工作流(共 %d 个节点)", len(execPayload.FlowContent.Nodes))})
execId, execErr = executeOrResume(progressCtx, conn, execPayload)
if errors.Is(execErr, errExecAlreadyRunning) {
return // 运行中/已触发恢复：不写终态（状态由后台恢复收敛）
}
// in-process 重试：程序报错且未耗尽 → retry_count++ 落库（保持 status=1，前端不闪失败）→ 续跑
retryCount := 0
for execErr != nil && execId > 0 && shouldRetry(execErr) && retryCount < execMaxRetryCount {
retryCount++
if err := sessionDao.ExecWorkflowDao.UpdateRetry(saveCtx, execId, 1, retryCount); err != nil {
glog.Errorf(saveCtx, "重试标记落库失败 execId=%d: %v", execId, err)
break
}
glog.Infof(saveCtx, "工作流执行失败，自动重试 %d/%d，execId=%d: %v", retryCount, execMaxRetryCount, execId, execErr)
_, execErr = reExecute(progressCtx, execId)
}
// 错误分类落库：用户取消 → retryable=0 不重试；
// 程序报错（非取消）→ retryable=1，retry_count 已记（重试循环内递增；UpdateRetry 失败 break 时也为当前值）
if execErr != nil {
if errors.Is(execErr, context.Canceled) {
_ = sessionDao.ExecWorkflowDao.UpdateRetry(saveCtx, execId, 0, 0)
} else {
_ = sessionDao.ExecWorkflowDao.UpdateRetry(saveCtx, execId, 1, retryCount)
if retryCount >= execMaxRetryCount {
// 终局清理（Task 5 裁定）：重试耗尽，exec 永久失败，清 flow_async_task 孤儿缓存
_ = flowDao.FlowAsyncTaskDao.DeleteByExecution(saveCtx, execId)
}
}
}
if !g.IsEmpty(execId) {
glog.Infof(saveCtx, "工作流执行完成，execId: %v", execId)
recordWorkflow(saveCtx, execId, time.Since(start), execErr)
@@ -283,6 +310,19 @@ func executeOrResume(ctx context.Context, conn *wsCommon.WsConnection, req *sess
return 0, fmt.Errorf("查询最近工作流执行记录失败: %v", err)
}
if lastExec != nil {
if *lastExec.Status == *flow.FlowExecutionStatusRunning.Code() {
// status=1：可能正在跑（本节点或其它节点后台恢复）或僵尸遗留。
// 心跳新鲜 → 真在跑，不新建避免双跑；心跳陈旧 → 僵尸，触发后台恢复。
// 都不新建执行，恢复在后台完成，完成后状态自然收敛（spec §9）。
nowMs := time.Now().UnixMilli()
if lastExec.LastHeartbeat < nowMs-int64(heartbeatStaleAfter/time.Millisecond) {
_ = writeJSON(conn, &wsCommon.WsPushMsg{Type: "round_start", Message: "检测到未完成执行，正在恢复", Data: map[string]interface{}{"id": lastExec.Id}})
go recoverExecution(context.WithoutCancel(ctx), lastExec.Id)
} else {
_ = writeJSON(conn, &wsCommon.WsPushMsg{Type: "round_start", Message: "工作流正在执行中", Data: map[string]interface{}{"id": lastExec.Id}})
}
return lastExec.Id, errExecAlreadyRunning
}
if *lastExec.Status == *flow.FlowExecutionStatusFailed.Code() && flowContentEqual(lastExec.RequestParams, req.FlowContent) {
_ = writeJSON(conn, &wsCommon.WsPushMsg{Type: "round_start", Message: "运行开始", Data: map[string]interface{}{
"id": lastExec.Id,
@@ -322,6 +362,7 @@ func execute(ctx context.Context, conn *wsCommon.WsConnection, execId int64, sta
NodeGroupId: nodeGroupId,
Status: flow.FlowExecutionStatusRunning.Code(),
RequestParams: req.FlowContent,
LastHeartbeat: time.Now().UnixMilli(),
})
if err != nil || g.IsEmpty(execId) {
glog.Errorf(ctx, "工作流执行记录创建失败: %v", err)
@@ -332,12 +373,12 @@ func execute(ctx context.Context, conn *wsCommon.WsConnection, execId int64, sta
// 需解引用按值比较，否则复用失败记录时不会重置为 Running、也不更新 RequestParams
if status != nil && *status == *flow.FlowExecutionStatusFailed.Code() {
glog.Infof(ctx, "工作流断点续跑execute，execId: %v", execId)
_, err = sessionDao.ExecWorkflowDao.Update(ctx, &sessionDto.UpdateWorkflowReq{
Id: execId,
NodeGroupId: nodeGroupId,
Status: flow.FlowExecutionStatusRunning.Code(),
RequestParams: req.FlowContent,
})
err = sessionDao.ExecWorkflowDao.ResetRunning(ctx, execId, nodeGroupId)
if err != nil {
return
}
// 复用失败记录时参数可能已变：把新参数落库，供后续 reExecute/恢复例程按记录参数续跑
_, err = sessionDao.ExecWorkflowDao.Update(ctx, &sessionDto.UpdateWorkflowReq{Id: execId, RequestParams: req.FlowContent})
if err != nil {
return
}
@@ -360,6 +401,8 @@ func execute(ctx context.Context, conn *wsCommon.WsConnection, execId int64, sta
_ = writeJSON(conn, &wsCommon.WsPushMsg{Type: "round_start", Message: "运行开始", Data: map[string]interface{}{
"id": execId,
}})
stop := startHeartbeat(ctx, execId)
defer stop()
err = BuildExecution(ctx, true, req.FlowId, execId, nodeGroupId, conn.SessionId, req.FlowContent)
if err != nil {
return
@@ -374,14 +417,12 @@ func reExecute(ctx context.Context, execWorkflowId int64) (id int64, err error)
return
}
var nodeGroupId = uuid.NewString()
_, err = sessionDao.ExecWorkflowDao.Update(ctx, &sessionDto.UpdateWorkflowReq{
Id: flowInfo.Id,
NodeGroupId: nodeGroupId,
Status: flow.FlowExecutionStatusRunning.Code(),
})
err = sessionDao.ExecWorkflowDao.ResetRunning(ctx, flowInfo.Id, nodeGroupId)
if err != nil {
return
}
stop := startHeartbeat(ctx, flowInfo.Id)
defer stop()
err = BuildExecution(ctx, false, flowInfo.FlowId, flowInfo.Id, nodeGroupId, flowInfo.SessionId, flowInfo.RequestParams)
if err != nil {
return
@@ -468,5 +509,7 @@ func BuildExecution(ctx context.Context, forceNewRun bool, flowId, executionId i
_ = flowDao.FlowCheckpointDao.Delete(ctx, gconv.String(executionId))
// 清理该执行段结果，下次执行无残留（失败则保留，供 reExecute 复用）
_ = flowDao.FlowSegmentResultDao.DeleteByExecution(ctx, executionId)
// 清理该执行异步任务缓存，下次执行无残留（失败则保留，供恢复复用）
_ = flowDao.FlowAsyncTaskDao.DeleteByExecution(ctx, executionId)
return
}