diff --git a/workflow/service/flow/flow_ws_exec.go b/workflow/service/flow/flow_ws_exec.go index a6f4e48..54d3770 100644 --- a/workflow/service/flow/flow_ws_exec.go +++ b/workflow/service/flow/flow_ws_exec.go @@ -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, diff --git a/workflow/service/session/session_service.go b/workflow/service/session/session_service.go index 23c722b..abb6274 100644 --- a/workflow/service/session/session_service.go +++ b/workflow/service/session/session_service.go @@ -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]++