fix: 修复工作流执行状态比较与落库逻辑

- 解引用状态指针按值比较,避免执行记录误显示为运行中
- panic 时兜底写入失败状态,防止记录卡在 Running
- 保存上下文提前至异步执行入口,确保断连后仍能落库
- 移除结果文件地址前缀拼接,使用原始 URL
This commit is contained in:
2026-08-22 08:28:04 +08:00
parent 5b93a7a507
commit aae4ebe199
2 changed files with 32 additions and 18 deletions
+21 -10
View File
@@ -118,17 +118,27 @@ func handleExecute(ctx context.Context, conn *wsCommon.WsConnection, payload int
// 异步执行工作流(直接 goroutine,不依赖上游 workerPool 二次排队)
go func() {
// 落库用不带取消的 ctx(保留 request 值),保证前端终止/断连后记录仍能写入
saveCtx := context.WithoutCancel(ctx)
start := time.Now()
var execId int64
var execErr error
recorded := false
defer func() {
if r := recover(); r != nil {
glog.Errorf(execCtx, "workflow panic: %v", r)
execErr = fmt.Errorf("工作流异常: %v", r)
// panic 被 recover 吞掉时,正常路径的 recordWorkflow 不会执行,
// 在此兜底把失败状态落库,避免 exec_workflow 记录卡在 Running
if !recorded && !g.IsEmpty(execId) {
recordWorkflow(saveCtx, execId, time.Since(start), execErr)
recorded = true
}
_ = writeJSON(conn, &wsCommon.WsPushMsg{Type: "error", Message: "工作流异常", Error: fmt.Sprintf("%v", r)})
}
}()
defer conn.SetMeta("execCancel", nil)
// 落库用不带取消的 ctx(保留 request 值),保证前端终止/断连后记录仍能写入
saveCtx := context.WithoutCancel(ctx)
// 会话落库:前端 sessionId 对应会话已存在则复用,否则按流程名新建
flowName := defaultSessionName
if flowUser, e := flowDao.FlowUserDao.Get(saveCtx, &flowDto.GetFlowUserReq{Id: execPayload.FlowId}); e == nil && flowUser != nil && flowUser.FlowName != "" {
@@ -139,20 +149,19 @@ func handleExecute(ctx context.Context, conn *wsCommon.WsConnection, payload int
_ = writeJSON(conn, &wsCommon.WsPushMsg{Type: "error", Message: "工作流会话创建失败", Error: fmt.Sprintf("%v", e)})
}
start := time.Now()
reporter := &wsProgressReporter{conn: conn}
progressCtx := context.WithValue(execCtx, wsProgressCtxKey{}, reporter)
_ = writeJSON(conn, &wsCommon.WsPushMsg{Type: "ack", Message: fmt.Sprintf("开始执行工作流(共 %d 个节点)", len(execPayload.FlowContent.Nodes))})
execId, err := executeOrResume(progressCtx, conn, execPayload)
execId, execErr = executeOrResume(progressCtx, conn, execPayload)
if !g.IsEmpty(execId) {
glog.Infof(saveCtx, "工作流执行完成,execId: %v", execId)
recordWorkflow(saveCtx, execId, time.Since(start), err)
recordWorkflow(saveCtx, execId, time.Since(start), execErr)
recorded = true
}
if err != nil {
_ = writeJSON(conn, &wsCommon.WsPushMsg{Type: "error", Message: "工作流执行失败", Error: err.Error()})
if execErr != nil {
_ = writeJSON(conn, &wsCommon.WsPushMsg{Type: "error", Message: "工作流执行失败", Error: execErr.Error()})
return
}
// 成功:把本次执行保存的结果文件路径(exec_workflow_result)一并推给前端
@@ -290,7 +299,9 @@ func execute(ctx context.Context, conn *wsCommon.WsConnection, execId int64, sta
return
}
} else {
if status == flow.FlowExecutionStatusFailed.Code() {
// FlowExecutionStatus 是 *int8 别名,Code() 返回包级指针,直接 == 是地址比较恒为 false,
// 需解引用按值比较,否则复用失败记录时不会重置为 Running、也不更新 RequestParams
if status != nil && *status == *flow.FlowExecutionStatusFailed.Code() {
_, err = sessionDao.ExecWorkflowDao.Update(ctx, &sessionDto.UpdateWorkflowReq{
Id: execId,
NodeGroupId: nodeGroupId,
+11 -8
View File
@@ -190,12 +190,16 @@ func chatExecVO(c *entity.ExecChat) *sessionDto.VOSessionInfoResult {
func workflowExecVO(w *entity.ExecWorkflow, resultFileUrl string) *sessionDto.VOSessionInfoResult {
status := 1
if w.Status == flow.FlowExecutionStatusFailed.Code() {
status = resultStatusFailed
} else if w.Status == flow.FlowExecutionStatusSuccess.Code() {
status = resultStatusSuccess
} else if w.Status == flow.FlowExecutionStatusCancel.Code() {
status = resultStatusCancel
// FlowExecutionStatus 是 *int8 别名,Code() 返回包级指针,直接 == 是地址比较恒为 false,
// 需解引用按值比较,否则所有执行记录在前端都会误显示为"运行中"
if w.Status != nil {
if *w.Status == *flow.FlowExecutionStatusFailed.Code() {
status = resultStatusFailed
} else if *w.Status == *flow.FlowExecutionStatusSuccess.Code() {
status = resultStatusSuccess
} else if *w.Status == *flow.FlowExecutionStatusCancel.Code() {
status = resultStatusCancel
}
}
return &sessionDto.VOSessionInfoResult{
Id: w.Id,
@@ -265,7 +269,6 @@ func (s *sessionService) ResultList(ctx context.Context, req *sessionDto.ListWor
execsByDate[date] = append(execsByDate[date], e)
}
prefix := res.ImgAddressPrefix
for _, d := range dates {
execList := execsByDate[d]
if len(execList) == 0 {
@@ -283,7 +286,7 @@ func (s *sessionService) ResultList(ctx context.Context, req *sessionDto.ListWor
if rf.ResultFileUrl == "" {
continue
}
content := prefix + rf.ResultFileUrl
content := rf.ResultFileUrl
ext := flowService.GetFileTypeByPath(content)
suffix := outputItemSuffix(ext)
suffixCount[suffix]++