585 lines
26 KiB
Go
585 lines
26 KiB
Go
package flow
|
||
|
||
import (
|
||
"ai-agent/workflow/consts/flow"
|
||
flowDao "ai-agent/workflow/dao/flow"
|
||
sessionDao "ai-agent/workflow/dao/session"
|
||
flowDto "ai-agent/workflow/model/dto/flow"
|
||
sessionDto "ai-agent/workflow/model/dto/session"
|
||
"ai-agent/workflow/model/entity"
|
||
"bytes"
|
||
"context"
|
||
"encoding/json"
|
||
"errors"
|
||
"fmt"
|
||
"time"
|
||
|
||
"gitea.redpowerfuture.com/red-future/common/utils"
|
||
wsCommon "gitea.redpowerfuture.com/red-future/common/websocket"
|
||
"github.com/cloudwego/eino/compose"
|
||
"github.com/gogf/gf/v2/frame/g"
|
||
"github.com/gogf/gf/v2/os/glog"
|
||
"github.com/gogf/gf/v2/util/gconv"
|
||
"github.com/google/uuid"
|
||
)
|
||
|
||
// ====================== WebSocket 服务器 ======================
|
||
|
||
func init() {
|
||
// 工作流消息处理器注册在统一的 SessionWsService（见 ws_server.go）上：
|
||
// 首次连接仅升级，连接后按消息 type 路由，不再建连时区分普通对话/工作流
|
||
SessionWsService.OnMessage("workflow", handleExecute)
|
||
SessionWsService.OnMessage("workflow_cancel", handleCancel)
|
||
}
|
||
|
||
// defaultSessionName 工作流执行但查不到流程名时，会话的兜底名称
|
||
const defaultSessionName = "工作流执行"
|
||
|
||
// errWorkflowTerminated 前端终止工作流执行时的错误标记（写入 exec_workflow.error_message）
|
||
var errWorkflowTerminated = "用户已终止执行"
|
||
|
||
// ====================== 进度上报 ======================
|
||
type wsProgressCtxKey struct{}
|
||
|
||
// ProgressReporter 节点执行进度回调接口
|
||
type ProgressReporter interface {
|
||
ReportStart(nodeId, nodeName string, nodeIndex, nodeCount int)
|
||
ReportComplete(nodeId, nodeName string, nodeIndex, nodeCount int)
|
||
}
|
||
|
||
// GetProgressReporter 从context中获取进度上报器
|
||
func GetProgressReporter(ctx context.Context) ProgressReporter {
|
||
if reporter, ok := ctx.Value(wsProgressCtxKey{}).(ProgressReporter); ok {
|
||
return reporter
|
||
}
|
||
return nil
|
||
}
|
||
|
||
// 进度上报由 exec_hub.go 的 execHub 实现（单执行事件中枢，可多连接订阅）；wsProgressCtxKey/ProgressReporter/GetProgressReporter 保留。
|
||
|
||
// ====================== 消息处理 ======================
|
||
|
||
// handleExecute 处理工作流执行（由 workerPool 异步调用，不阻塞读循环）
|
||
func handleExecute(ctx context.Context, conn *wsCommon.WsConnection, payload interface{}) {
|
||
execPayload := new(sessionDto.WebSocketExecWorkflowReq)
|
||
if err := gconv.Struct(payload, execPayload); err != nil {
|
||
glog.Errorf(ctx, "工作流执行参数解析失败: %v", err)
|
||
_ = writeJSON(conn, &wsCommon.WsPushMsg{Type: "error", Message: "执行参数解析失败", Error: err.Error()})
|
||
return
|
||
}
|
||
|
||
execCtx, execCancel := context.WithCancel(ctx)
|
||
|
||
// 替换旧 cancel：同一连接上重新执行时取消上一次遗留的执行
|
||
if oldCancel := getExecCancel(conn); oldCancel != nil {
|
||
oldCancel()
|
||
}
|
||
|
||
// 异步执行工作流（直接 goroutine，不依赖上游 workerPool 二次排队）
|
||
go func() {
|
||
// 事件中枢：同 session+flow 已有运行中执行（本进程）则复用其 hub，让本连接附着其进度；
|
||
// 否则新建注册为候选，由本执行（execute/reExecute）或恢复例程接管
|
||
hub, _ := registerHubIfAbsent(conn.SessionId, execPayload.FlowId, newExecHub(conn.SessionId, execPayload.FlowId))
|
||
if !hub.Owned() {
|
||
// 候选/未持有：预留取消为当前连接的 execCancel（新执行 MarkOwned 后生效）；
|
||
// 已持有（同 session+flow 运行中执行）则保留持有者 cancel，避免覆盖导致取消错对象
|
||
hub.SetCancel(execCancel)
|
||
}
|
||
subscribeConnToHub(conn, hub)
|
||
progressCtx := context.WithValue(execCtx, wsProgressCtxKey{}, hub)
|
||
|
||
// 登记运行中执行：优雅关停（SetShuttingDown）统一取消本执行；finish 在落完终态后解除登记，
|
||
// main 的 WaitExecRunsDrain 据此等待 WS 执行落库后再退出（避免进程退出时记录仍卡 status=1）
|
||
finish := trackExecRun(execCancel)
|
||
defer finish()
|
||
// 落库用不带取消的 ctx（保留 request 值），保证前端终止/断连后记录仍能写入
|
||
saveCtx := context.WithoutCancel(ctx)
|
||
start := time.Now()
|
||
var execId int64
|
||
var execErr error
|
||
recorded := false
|
||
owner := false // 本 goroutine 是否为执行持有者（附着/触发恢复时不持有）
|
||
// 持有者收尾：落完终态广播后关闭 hub（退订全部连接/清 meta/注销）。
|
||
// 附着路径不置 owner，由运行中的持有方统一 Close；panic 路径在下方 recover defer 中置 owner 兜底。
|
||
// 先注册此 defer → 后注册的 panic defer 先执行（先广播再 Close）
|
||
defer func() {
|
||
if owner {
|
||
hub.Close()
|
||
}
|
||
}()
|
||
defer func() {
|
||
if r := recover(); r != nil {
|
||
glog.Errorf(execCtx, "workflow panic: %v", r)
|
||
execErr = fmt.Errorf("工作流异常: %v", r)
|
||
// panic 发生在 executeOrResume 内部（如节点 lambda panic）时，
|
||
// 多返回值赋值不会完成，此处 execId 可能为 0，需走兜底按会话+流程查最近"运行中"记录标记失败，
|
||
// 避免 exec_workflow 记录卡在 Running
|
||
if !recorded {
|
||
recordExecutionFailure(saveCtx, conn.SessionId, execPayload.FlowId, execId, execErr)
|
||
recorded = true
|
||
}
|
||
// panic 必然发生在本 goroutine 持有的执行内（附着/恢复路径不跑 BuildExecution，不会 panic）
|
||
owner = true
|
||
hub.Publish(&wsCommon.WsPushMsg{Type: "error", Message: "工作流异常", Error: fmt.Sprintf("%v", r)})
|
||
}
|
||
}()
|
||
|
||
// 会话落库：前端 sessionId 对应会话已存在则复用，否则按流程名新建
|
||
flowName := defaultSessionName
|
||
if flowUser, e := flowDao.FlowUserDao.Get(saveCtx, &flowDto.GetFlowUserReq{Id: execPayload.FlowId}); e == nil && flowUser != nil && flowUser.FlowName != "" {
|
||
flowName = flowUser.FlowName
|
||
}
|
||
if e := ensureSession(saveCtx, conn.SessionId, flowName); e != nil {
|
||
glog.Errorf(saveCtx, "工作流会话创建失败: %v", e)
|
||
_ = writeJSON(conn, &wsCommon.WsPushMsg{Type: "error", Message: "工作流会话创建失败", Error: fmt.Sprintf("%v", e)})
|
||
}
|
||
|
||
_ = 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) {
|
||
// 已附着到运行中执行 / 已触发恢复 / 其它节点在跑：
|
||
// 连接已订阅到 hub（运行中持有者广播进度与终态并统一 Close；executeOrResume 对占位 hub 已 Close）。
|
||
// 本 goroutine 不落终态、不 Close（owner=false），由持有方收敛。
|
||
return
|
||
}
|
||
owner = true
|
||
if !g.IsEmpty(execId) {
|
||
hub.execId = execId
|
||
}
|
||
// 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)
|
||
// 进程内重试：exec 仍为本进程持有的 status=1 运行中，条件重置仅允许从运行中重置
|
||
_, execErr = reExecute(progressCtx, execId, *flow.FlowExecutionStatusRunning.Code())
|
||
if errors.Is(execErr, errExecAlreadyRunning) {
|
||
return // 状态已被其它路径抢占：不写终态，由对方收敛
|
||
}
|
||
}
|
||
// 错误分类：程序关停取消 → retryable=1（下次启动恢复续跑）；
|
||
// 用户取消 → retryable=0 不重试；
|
||
// 程序报错（非取消）→ retryable=1，retry_count 已记（重试循环内递增）
|
||
// retryable/retry_count 随终态 recordWorkflow 一次原子写入，不单独 UpdateRetry
|
||
var retryable, retryCnt *int
|
||
if execErr != nil {
|
||
switch {
|
||
case errors.Is(execErr, context.Canceled) && IsShuttingDown():
|
||
// 关停标记置位：连接 ctx 取消来自程序优雅关停，非用户取消，
|
||
// 换错误标记让 recordWorkflow 写"程序关停中断"，并可重试以便下次启动恢复
|
||
retryable, retryCnt = intptr(1), intptr(retryCount)
|
||
execErr = errInterruptedByShutdown
|
||
case errors.Is(execErr, context.Canceled):
|
||
retryable, retryCnt = intptr(0), intptr(0)
|
||
default:
|
||
retryable, retryCnt = intptr(1), intptr(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, retryable, retryCnt)
|
||
recorded = true
|
||
} else if execErr != nil {
|
||
// 查询/创建执行记录失败（拿不到 execId）时，兜底把该会话+工作流最近一条"运行中"记录标记为失败
|
||
recordExecutionFailure(saveCtx, conn.SessionId, execPayload.FlowId, execId, execErr)
|
||
recorded = true
|
||
}
|
||
if execErr != nil {
|
||
// 终态广播（发起连接 + 附着订阅者）
|
||
hub.Publish(&wsCommon.WsPushMsg{Type: "error", Message: "工作流执行失败", Error: execErr.Error()})
|
||
return
|
||
}
|
||
// 成功：把本次执行保存的结果文件路径（exec_workflow_result）一并推给前端
|
||
hub.Publish(&wsCommon.WsPushMsg{
|
||
Type: "flow_complete",
|
||
Message: "工作流执行完成",
|
||
Data: map[string]interface{}{
|
||
"resultFileUrls": workflowResultFileUrls(saveCtx, execId),
|
||
},
|
||
})
|
||
}()
|
||
}
|
||
|
||
// recordExecutionFailure 记录一次失败状态。有 execId 直接更新该记录；拿不到 execId
|
||
// （executeOrResume 在创建记录后、返回前 panic，或查询/创建执行记录失败）时，
|
||
// 兜底按会话+工作流查最近一条仍处于"运行中"的记录标记为失败，避免前端已报错但记录卡在 Running。
|
||
// 最近记录已是成功/失败状态则不处理（可能是上一次执行的结果，不应误改）。
|
||
func recordExecutionFailure(ctx context.Context, sessionId string, flowId int64, execId int64, runErr error) {
|
||
// 兜底失败路径同样携带 retryable 分类（与 handleExecute 一致）：
|
||
// 用户取消=0；其余（程序关停中断/程序报错）可重试=1，保证任意 status=3 写库都带分类
|
||
var retryable, retryCnt *int
|
||
if runErr != nil {
|
||
if errors.Is(runErr, context.Canceled) && !IsShuttingDown() {
|
||
retryable, retryCnt = intptr(0), intptr(0)
|
||
} else {
|
||
retryable, retryCnt = intptr(1), intptr(0)
|
||
}
|
||
}
|
||
if !g.IsEmpty(execId) {
|
||
recordWorkflow(ctx, execId, 0, runErr, retryable, retryCnt)
|
||
return
|
||
}
|
||
lastExec, err := sessionDao.ExecWorkflowDao.GetLatestBySessionAndFlow(ctx, sessionId, flowId)
|
||
if err != nil || lastExec == nil {
|
||
glog.Errorf(ctx, "兜底标记失败状态失败: sessionId=%s flowId=%d err=%v", sessionId, flowId, err)
|
||
return
|
||
}
|
||
if lastExec.Status == nil || *lastExec.Status != *flow.FlowExecutionStatusRunning.Code() {
|
||
return
|
||
}
|
||
recordWorkflow(ctx, lastExec.Id, 0, runErr, retryable, retryCnt)
|
||
}
|
||
|
||
// intptr 取 int 值指针（recordWorkflow 的 retryable/retryCount 参数：nil=不改动）
|
||
func intptr(v int) *int { return &v }
|
||
|
||
// recordWorkflow 把一次工作流执行写入 exec_workflow/exec_workflow_result：运行记录 + 输出文件结果。
|
||
// retryable/retryCount 非 nil 时随终态同语句原子落库，避免"先 UpdateRetry 再 Update"两步写部分生效
|
||
// 导致 retryable 与 status/error_message 不一致（关停中断场景曾出现 retryable=0 但 message=程序关停中断）
|
||
func recordWorkflow(ctx context.Context, id int64, duration time.Duration, runErr error, retryable, retryCount *int) {
|
||
// exec_workflow 状态沿用 1-运行中,2-成功,3-失败；前端结果卡片也只识别 1/2/3
|
||
// （4 会误显示为"运行中"），故取消同样记为失败，错误信息写"用户已终止执行"
|
||
// error_message 存友好提示，error 存原始错误明细
|
||
status := flow.FlowExecutionStatusSuccess
|
||
var errorMessage, errorDetail string
|
||
if runErr != nil {
|
||
status = flow.FlowExecutionStatusFailed
|
||
switch {
|
||
case errors.Is(runErr, context.Canceled):
|
||
errorMessage = errWorkflowTerminated
|
||
case errors.Is(runErr, errInterruptedByShutdown):
|
||
errorMessage = "程序关停中断"
|
||
default:
|
||
errorMessage = "工作流执行失败"
|
||
errorDetail = runErr.Error()
|
||
}
|
||
}
|
||
data := map[string]any{
|
||
entity.ExecWorkflowCol.Status: *status.Code(),
|
||
}
|
||
if d := int64(duration.Seconds()); d != 0 {
|
||
data[entity.ExecWorkflowCol.Duration] = d
|
||
}
|
||
if errorMessage != "" {
|
||
data[entity.ExecWorkflowCol.ErrorMessage] = errorMessage
|
||
}
|
||
if errorDetail != "" {
|
||
data[entity.ExecWorkflowCol.Error] = errorDetail
|
||
}
|
||
if retryable != nil {
|
||
data[entity.ExecWorkflowCol.Retryable] = *retryable
|
||
data[entity.ExecWorkflowCol.RetryCount] = *retryCount
|
||
}
|
||
if err := sessionDao.ExecWorkflowDao.UpdateMap(ctx, id, data); err != nil {
|
||
glog.Errorf(ctx, "exec_workflow 终态落库失败 execId=%d: %v", id, err)
|
||
return
|
||
}
|
||
// 执行成功：重新执行复用了同一条记录，需显式清空，避免上一次失败的报错残留
|
||
if runErr == nil {
|
||
if _, err := sessionDao.ExecWorkflowDao.ClearError(ctx, id); err != nil {
|
||
glog.Errorf(ctx, "exec_workflow 报错信息清空失败: %v", err)
|
||
}
|
||
}
|
||
}
|
||
|
||
// workflowResultFileUrls 查询指定工作流执行保存的结果文件路径（带文件前缀，与 session/get 返回一致）
|
||
func workflowResultFileUrls(ctx context.Context, execId int64) []string {
|
||
results, err := sessionDao.ExecWorkflowResultDao.ListByExecId(ctx, execId)
|
||
if err != nil {
|
||
glog.Errorf(ctx, "查询工作流结果路径失败: %v", err)
|
||
return nil
|
||
}
|
||
prefix, _ := utils.GetFileAddressPrefix(ctx)
|
||
urls := make([]string, 0, len(results))
|
||
for _, r := range results {
|
||
if r.ResultFileUrl != "" {
|
||
urls = append(urls, prefix+r.ResultFileUrl)
|
||
}
|
||
}
|
||
return urls
|
||
}
|
||
|
||
// handleCancel 取消工作流执行
|
||
func handleCancel(ctx context.Context, conn *wsCommon.WsConnection, _ interface{}) {
|
||
if cancel := getExecCancel(conn); cancel != nil {
|
||
cancel()
|
||
}
|
||
_ = writeJSON(conn, &wsCommon.WsPushMsg{Type: "ack", Message: "已取消工作流执行"})
|
||
}
|
||
|
||
// ====================== 工具函数 ======================
|
||
|
||
func getExecCancel(conn *wsCommon.WsConnection) context.CancelFunc {
|
||
cancel, _ := wsCommon.GetMetaT[context.CancelFunc](conn, "execCancel")
|
||
return cancel
|
||
}
|
||
|
||
// writeJSON 业务层写入，委托 WsConnection.WriteJSON（共享 writeMu 写锁）
|
||
func writeJSON(conn *wsCommon.WsConnection, data interface{}) error {
|
||
return conn.WriteJSON(data)
|
||
}
|
||
|
||
// executeOrResume 决策工作流执行方式：
|
||
// - 同会话+同工作流的最近一次执行失败，且本次传递参数与上次一致 → 断点续跑（reExecute，复用原执行记录，从失败断点继续）
|
||
// - 其余情况（上次成功 / 上次参数与本次不同 / 无历史记录 / 查询出错）→ 全新执行（execute）
|
||
func executeOrResume(ctx context.Context, conn *wsCommon.WsConnection, req *sessionDto.WebSocketExecWorkflowReq) (id int64, err error) {
|
||
lastExec, err := sessionDao.ExecWorkflowDao.GetLatestBySessionAndFlow(ctx, conn.SessionId, req.FlowId)
|
||
if err != nil {
|
||
glog.Errorf(ctx, "查询最近工作流执行记录失败: %v", err)
|
||
return 0, fmt.Errorf("查询最近工作流执行记录失败: %v", err)
|
||
}
|
||
if lastExec != nil {
|
||
if *lastExec.Status == *flow.FlowExecutionStatusRunning.Code() {
|
||
// status=1：可能正在跑（本节点或其它节点后台恢复）或僵尸遗留。
|
||
// 心跳新鲜 → 真在跑，不新建避免双跑；心跳陈旧 → 僵尸，触发后台恢复。
|
||
// 都不新建执行，恢复在后台完成，完成后状态自然收敛（spec §9）。
|
||
nowMs := time.Now().UnixMilli()
|
||
hub := getProgressHub(ctx)
|
||
owned := hub != nil && hub.Owned()
|
||
if lastExec.LastHeartbeat < nowMs-int64(heartbeatStaleAfter/time.Millisecond) {
|
||
// 心跳陈旧 = 僵尸遗留：本进程恢复例程正在跑则附着其进度；否则拉起恢复并附着。
|
||
// 候选 hub 直接交给恢复例程复用同一实例（不 close）：最终执行者与连接共用一 hub，
|
||
// 取消/进度不因"占位被关→重建"竞态丢失
|
||
if !owned {
|
||
go recoverExecution(context.WithoutCancel(ctx), lastExec.Id, &execAttach{conn: conn, sessionId: conn.SessionId, flowId: req.FlowId, hub: hub})
|
||
}
|
||
_ = writeJSON(conn, &wsCommon.WsPushMsg{Type: "round_start", Message: "检测到未完成执行，正在恢复", Data: map[string]interface{}{"id": lastExec.Id}})
|
||
} else {
|
||
// 心跳新鲜 = 真在跑：owned → 连接已附着本进程运行中执行，直接订阅其进度；
|
||
// !owned → 其它节点在跑（无法跨节点附着）或本进程执行尚未接管（候选 hub 会被其接管）。
|
||
// 不 close 候选：避免在 execute()/恢复 TryOwn 前误关，导致执行者进度/取消无人接收
|
||
_ = 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,
|
||
}})
|
||
glog.Infof(ctx, "工作流断点续跑，execId: %v", lastExec.Id)
|
||
return reExecute(ctx, lastExec.Id, *flow.FlowExecutionStatusFailed.Code())
|
||
}
|
||
glog.Infof(ctx, "工作流全新执行，lastExec: %v", lastExec)
|
||
return execute(ctx, conn, lastExec.Id, lastExec.Status, req)
|
||
}
|
||
glog.Infof(ctx, "工作流全新执行，无历史记录")
|
||
return execute(ctx, conn, 0, nil, req)
|
||
}
|
||
|
||
// flowContentEqual 判断两次工作流参数是否一致（JSON 序列化后字节比对。
|
||
// Go struct 按字段声明序序列化、map 键自动排序，同一内容结果确定，可用于参数等价判断）
|
||
func flowContentEqual(a, b *entity.FlowInfo) bool {
|
||
if a == nil || b == nil {
|
||
return a == b
|
||
}
|
||
ab, err1 := json.Marshal(a)
|
||
bb, err2 := json.Marshal(b)
|
||
if err1 != nil || err2 != nil {
|
||
return false
|
||
}
|
||
return bytes.Equal(ab, bb)
|
||
}
|
||
|
||
// execute 执行工作流（首次执行；同会话+同工作流最近一次执行为失败状态时复用该记录重新执行，不新建数据）
|
||
func execute(ctx context.Context, conn *wsCommon.WsConnection, execId int64, status flow.FlowExecutionStatus, req *sessionDto.WebSocketExecWorkflowReq) (id int64, err error) {
|
||
// 优雅关停期间不再启动新执行（避免关停后仍登记运行、落库被退出进程打断；走 Canceled 分类落终态）
|
||
if IsShuttingDown() {
|
||
return 0, context.Canceled
|
||
}
|
||
var nodeGroupId = uuid.NewString()
|
||
if g.IsEmpty(execId) {
|
||
glog.Infof(ctx, "工作流全新执行execute，无历史记录")
|
||
execId, err = sessionDao.ExecWorkflowDao.Insert(ctx, &sessionDto.CreateWorkflowReq{
|
||
SessionId: conn.SessionId,
|
||
FlowId: req.FlowId,
|
||
NodeGroupId: nodeGroupId,
|
||
Status: flow.FlowExecutionStatusRunning.Code(),
|
||
RequestParams: req.FlowContent,
|
||
LastHeartbeat: time.Now().UnixMilli(),
|
||
})
|
||
if err != nil || g.IsEmpty(execId) {
|
||
glog.Errorf(ctx, "工作流执行记录创建失败: %v", err)
|
||
return
|
||
}
|
||
} else {
|
||
// FlowExecutionStatus 是 *int8 别名，Code() 返回包级指针，直接 == 是地址比较恒为 false，
|
||
// 需解引用按值比较，否则复用失败记录时不会重置为 Running、也不更新 RequestParams
|
||
if status != nil && *status == *flow.FlowExecutionStatusFailed.Code() {
|
||
glog.Infof(ctx, "工作流断点续跑execute，execId: %v", execId)
|
||
var reset bool
|
||
reset, err = sessionDao.ExecWorkflowDao.ResetRunning(ctx, execId, nodeGroupId, *flow.FlowExecutionStatusFailed.Code())
|
||
if err != nil {
|
||
return
|
||
}
|
||
if !reset {
|
||
// 已被其它路径（恢复例程/并发触发）抢先重置为运行中：放弃本次执行，状态由持有方收敛
|
||
return execId, errExecAlreadyRunning
|
||
}
|
||
// 复用失败记录时参数可能已变：把新参数落库，供后续 reExecute/恢复例程按记录参数续跑
|
||
_, err = sessionDao.ExecWorkflowDao.Update(ctx, &sessionDto.UpdateWorkflowReq{Id: execId, RequestParams: req.FlowContent})
|
||
if err != nil {
|
||
return
|
||
}
|
||
} else {
|
||
glog.Infof(ctx, "工作流全新执行execute，lastExec: %v", execId)
|
||
execId, err = sessionDao.ExecWorkflowDao.Insert(ctx, &sessionDto.CreateWorkflowReq{
|
||
SessionId: conn.SessionId,
|
||
FlowId: req.FlowId,
|
||
NodeGroupId: nodeGroupId,
|
||
Status: flow.FlowExecutionStatusRunning.Code(),
|
||
RequestParams: req.FlowContent,
|
||
LastHeartbeat: time.Now().UnixMilli(),
|
||
})
|
||
if err != nil || g.IsEmpty(execId) {
|
||
glog.Errorf(ctx, "工作流执行记录创建失败: %v", err)
|
||
return
|
||
}
|
||
}
|
||
}
|
||
|
||
_ = writeJSON(conn, &wsCommon.WsPushMsg{Type: "round_start", Message: "运行开始", Data: map[string]interface{}{
|
||
"id": execId,
|
||
}})
|
||
// WS 路径心跳与 BuildExecution 已共享同一 ctx（连接取消/DB 故障同生共死），无需租约丢失回调
|
||
stop := startHeartbeat(ctx, execId, nil)
|
||
defer stop()
|
||
// 接管候选 hub：本 goroutine 确认真正启动执行前标记 owned，供并发附着连接识别运行中持有者
|
||
if h := getProgressHub(ctx); h != nil && !h.Owned() {
|
||
h.MarkOwned()
|
||
}
|
||
err = BuildExecution(ctx, true, req.FlowId, execId, nodeGroupId, conn.SessionId, req.FlowContent)
|
||
if err != nil {
|
||
return
|
||
}
|
||
return execId, nil
|
||
}
|
||
|
||
// reExecute 重新执行工作流。
|
||
// prevStatus：调用方当前观察到的记录状态（失败=3 断点续跑；运行中=1 本进程自动重试），
|
||
// 传给 ResetRunning 做条件重置：仅当记录仍处于该状态时才重置（原子防双跑）。
|
||
// 状态已被其它路径抢先变更时返回 errExecAlreadyRunning，外层不写终态、由持有方收敛。
|
||
func reExecute(ctx context.Context, execWorkflowId int64, prevStatus int8) (id int64, err error) {
|
||
// 优雅关停期间不再启动新执行（返回 Canceled 让外层分类落 errInterruptedByShutdown 终态）
|
||
if IsShuttingDown() {
|
||
return 0, context.Canceled
|
||
}
|
||
flowInfo, err := sessionDao.ExecWorkflowDao.GetById(ctx, execWorkflowId)
|
||
if err != nil {
|
||
return
|
||
}
|
||
var nodeGroupId = uuid.NewString()
|
||
reset, err := sessionDao.ExecWorkflowDao.ResetRunning(ctx, flowInfo.Id, nodeGroupId, prevStatus)
|
||
if err != nil {
|
||
return
|
||
}
|
||
if !reset {
|
||
// 状态已被其它路径（恢复例程/并发触发）抢先重置为运行中：放弃续跑
|
||
return flowInfo.Id, errExecAlreadyRunning
|
||
}
|
||
// WS 路径心跳与 BuildExecution 已共享同一 ctx（连接取消/DB 故障同生共死），无需租约丢失回调
|
||
stop := startHeartbeat(ctx, flowInfo.Id, nil)
|
||
defer stop()
|
||
// 接管候选 hub：确认续跑启动前标记 owned（与 execute 一致）
|
||
if h := getProgressHub(ctx); h != nil && !h.Owned() {
|
||
h.MarkOwned()
|
||
}
|
||
err = BuildExecution(ctx, false, flowInfo.FlowId, flowInfo.Id, nodeGroupId, flowInfo.SessionId, flowInfo.RequestParams)
|
||
if err != nil {
|
||
return
|
||
}
|
||
return flowInfo.Id, nil
|
||
}
|
||
|
||
func BuildExecution(ctx context.Context, forceNewRun bool, flowId, executionId int64, nodeGroupId string, sessionId string, flowContent *entity.FlowInfo) (err error) {
|
||
// =========================================================================
|
||
// 构建执行图
|
||
// =========================================================================
|
||
var nodeList []entity.FlowNode
|
||
var runGraph compose.Runnable[any, any]
|
||
nodeList, runGraph, err = BuildGraphFromFlowContent(ctx, flowContent)
|
||
if err != nil {
|
||
return fmt.Errorf("执行工作流失败: %v", err)
|
||
}
|
||
|
||
// =========================================================================
|
||
// 构建 ConfigMap
|
||
// =========================================================================
|
||
nodeInputParams := ExtractFlowNodeFrom(flowContent)
|
||
configMap := make(map[string]*entity.FlowNode)
|
||
for _, cfg := range nodeInputParams {
|
||
configMap[cfg.Id] = cfg
|
||
}
|
||
for _, i := range nodeList {
|
||
configMap[i.Id] = &i
|
||
}
|
||
|
||
// =========================================================================
|
||
// 构建全局执行入参
|
||
// =========================================================================
|
||
execInput := &flowDto.FlowExecutionInput{
|
||
NodeGroupId: nodeGroupId,
|
||
ExecutionId: executionId,
|
||
FlowId: flowId,
|
||
ConfigMap: configMap,
|
||
SessionId: sessionId,
|
||
ForceNewRun: forceNewRun,
|
||
}
|
||
|
||
var opts []compose.Option
|
||
opts = append(opts, compose.WithCheckPointID(gconv.String(executionId)))
|
||
if forceNewRun {
|
||
opts = append(opts, compose.WithForceNewRun())
|
||
}
|
||
// 全新执行前清理该执行残留段结果：forceNewRun 复用同一条 exec 记录时可能留有旧参数生成的段，
|
||
// 不清则续跑会误复用。只按 execution_id 清理（放在图启动前，避免多视频节点互相误删）
|
||
if forceNewRun {
|
||
if err := flowDao.FlowSegmentResultDao.DeleteByExecution(ctx, executionId); err != nil {
|
||
return fmt.Errorf("清理段结果失败: %v", err)
|
||
}
|
||
}
|
||
_, err = runGraph.Invoke(ctx, execInput, opts...)
|
||
if err != nil {
|
||
// 图执行被 ctx 取消（WS 断连/用户终止）：返回 context.Canceled 语义，让 recordWorkflow
|
||
// 记为"用户已终止执行"。此时 Eino 已把断点写入 checkpoint store（DbCheckPointStore 用
|
||
// WithoutCancel 落库），重新提交相同参数即可断点续跑。
|
||
if ctxErr := ctx.Err(); ctxErr != nil {
|
||
return fmt.Errorf("执行工作流失败: %w", ctxErr)
|
||
}
|
||
info, infoOk := compose.ExtractInterruptInfo(err)
|
||
if infoOk {
|
||
var errMsg string
|
||
var errNodeCount int
|
||
for _, item := range info.InterruptContexts {
|
||
if item.Info == nil {
|
||
continue
|
||
}
|
||
if g.NewVar(item.Info).IsMap() {
|
||
errNodeCount++
|
||
valMap := gconv.Map(item.Info)
|
||
errMsg = fmt.Sprintf("%v\n%v", errMsg, fmt.Sprintf("节点：%v, 失败原因：%v", valMap["node"], valMap["error"]))
|
||
}
|
||
}
|
||
if !g.IsEmpty(errMsg) {
|
||
err = fmt.Errorf("%v个节点，%v", errNodeCount, errMsg)
|
||
}
|
||
}
|
||
return fmt.Errorf("执行工作流失败: %v", err)
|
||
}
|
||
// 清理断点数据
|
||
_ = flowDao.FlowCheckpointDao.Delete(ctx, gconv.String(executionId))
|
||
// 清理该执行段结果，下次执行无残留（失败则保留，供 reExecute 复用）
|
||
_ = flowDao.FlowSegmentResultDao.DeleteByExecution(ctx, executionId)
|
||
// 清理该执行异步任务缓存，下次执行无残留（失败则保留，供恢复复用）
|
||
_ = flowDao.FlowAsyncTaskDao.DeleteByExecution(ctx, executionId)
|
||
return
|
||
}
|