19904408334
73f296731c
将 checkpoint、异步任务与段结果缓存的唯一键从 execution_id 改为 node_group_id，区分换参重跑与续跑语义；恢复/续跑复用执行记录的组标识，换参重跑则换新组并软删旧组残留，消除软删墓碑导致同键重存失效的问题。
539 lines
28 KiB
Go
539 lines
28 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"
|
||
"strings"
|
||
"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/net/ghttp"
|
||
"github.com/gogf/gf/v2/os/glog"
|
||
"github.com/gogf/gf/v2/util/gconv"
|
||
"github.com/google/uuid"
|
||
)
|
||
|
||
// ====================== WebSocket 服务器 ======================
|
||
|
||
func init() {
|
||
// 工作流消息处理器注册在统一的 SessionWsService 上：
|
||
// 首次连接仅升级，连接后按消息 type 路由，不再建连时区分普通对话/工作流
|
||
SessionWsService.OnMessage("workflow", handleExecute)
|
||
SessionWsService.OnMessage("workflow_cancel", handleCancel)
|
||
}
|
||
|
||
// defaultSessionName 工作流执行但查不到流程名时，会话的兜底名称
|
||
const defaultSessionName = "工作流执行"
|
||
|
||
// errWorkflowTerminated 前端终止工作流执行时的错误标记（写入 exec_workflow.error_message）
|
||
var errWorkflowTerminated = "用户已终止执行"
|
||
|
||
// ====================== 消息处理 ======================
|
||
|
||
// 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)
|
||
|
||
// 同流程运行中再点"执行" = 附着看进度（此处不做预取消）：下方 registerHubIfAbsent 复用同
|
||
// session+flow 的运行中 hub，executeOrResume 返回 errExecAlreadyRunning，本连接附着其进度/取消，
|
||
// 不新建执行。原 getExecCancel 无条件预取消会把运行中的同流程执行一并杀掉，与附着语义冲突，故弃用。
|
||
// 仅当本连接 meta execHub 指向的是另一流程（换流程执行）的运行中 hub 时终止它：
|
||
// 避免会话内串跑两个流程、旧流程在后台继续消耗计费（workflow_cancel 是显式取消入口）。
|
||
if prev, ok := wsCommon.GetMetaT[*execHub](conn, "execHub"); ok && prev != nil && prev.flowId != execPayload.FlowId {
|
||
prev.CancelByUser()
|
||
}
|
||
|
||
// 异步执行工作流（直接 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
|
||
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)
|
||
// 只置错误与所有权，不做终态落库/广播：recover 后本 goroutine 从 panic 处（execId/execErr
|
||
// 赋值可能未完成）继续执行，下方常规错误路径恰好覆盖此场景且只走一遍——
|
||
// execId=0 → else-if 兜底 recordExecutionFailure（按会话+流程查最近 Running 标记失败）；
|
||
// execId>0 → recordWorkflow；随后单次终态 Publish。若在此提前落库/广播，主路径会重复
|
||
// 一次（前端连收两条错误、兜底标记被调两遍）。
|
||
// panic 必然发生在本 goroutine 持有的执行内（附着/恢复路径不跑 BuildExecution，不会 panic）
|
||
owner = true
|
||
}
|
||
}()
|
||
|
||
// 会话落库：前端 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)})
|
||
}
|
||
|
||
// flowContent 经 WS 的 gconv 解析不跑 v:"required" 校验，缺省时按 0 节点提示，不 panic
|
||
nodeCount := 0
|
||
if execPayload.FlowContent != nil {
|
||
nodeCount = len(execPayload.FlowContent.Nodes)
|
||
}
|
||
_ = writeJSON(conn, &wsCommon.WsPushMsg{Type: "ack", Message: fmt.Sprintf("开始执行工作流(共 %d 个节点)", nodeCount)})
|
||
|
||
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)
|
||
case errors.Is(execErr, errBillingGateBlocked):
|
||
// 计费门禁拦截：余额不足/钱包不可用/费率非法，终局失败不重试不恢复
|
||
retryable, retryCnt = intptr(0), intptr(0)
|
||
default:
|
||
// 程序报错且重试耗尽 → exec 永久失败：async/segment/checkpoint 一律**保留**，
|
||
// 供用户同参数再点续跑（reExecute）复用已产出、免重复调模型/免双扣；
|
||
// 换参数重跑走 execute→forceNewRun，BuildExecution 起跑前统一清空，不误用旧残留
|
||
retryable, retryCnt = intptr(1), intptr(retryCount)
|
||
}
|
||
}
|
||
if !g.IsEmpty(execId) {
|
||
glog.Infof(saveCtx, "工作流执行完成，execId: %v", execId)
|
||
recordWorkflow(saveCtx, execId, time.Since(start), execErr, retryable, retryCnt)
|
||
} else if execErr != nil {
|
||
// 查询/创建执行记录失败（拿不到 execId）时，兜底把该会话+工作流最近一条"运行中"记录标记为失败
|
||
recordExecutionFailure(saveCtx, conn.SessionId, execPayload.FlowId, execId, execErr)
|
||
}
|
||
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),
|
||
},
|
||
})
|
||
}()
|
||
}
|
||
|
||
// executeOrResume 是"并发触发仲裁"的用户侧决策点：同一条 exec 的多个触发方
|
||
// （自动重试 / 手动续跑 / 恢复扫描 / 用户点击）同时发生时，谁真正跑由条件重置
|
||
// （ResetRunning/ResetRunningIfRecoverable，DB 行级原子）定夺，输家一律附着观察或
|
||
// 放弃（errExecAlreadyRunning），绝不双跑。点击落点分情形（status=1 心跳新鲜→附着 /
|
||
// 陈旧→触发恢复 / status=3 同参→手动续跑 / 其余→execute）见根目录
|
||
// 《工作流执行并发仲裁设计.md》§1/§4。
|
||
//
|
||
// 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：可能正在跑（本节点或其它节点后台恢复）或僵尸遗留。
|
||
// 心跳新鲜 → 真在跑，不新建避免双跑；心跳陈旧 → 僵尸，触发后台恢复。
|
||
// 都不新建执行：恢复在后台完成，完成后状态自然收敛——输家不写终态、由持有方收敛
|
||
// （仲裁语义见《工作流执行并发仲裁设计.md》§1/§4）
|
||
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()
|
||
// 记录发起执行用户的数字 ID：崩溃恢复续跑无 WS/HTTP 用户，需按 exec.user_id 补全合成用户
|
||
// 的 Id，外发 model-gateway/modelCall 的 X-User-Info 才能过单次调用最低余额门禁
|
||
// （creator 仅存 userName，推不回数字 id）。取不到用户不阻塞执行（openBillingOrder 后续会拦）；
|
||
// user_id=0 仅影响此类记录自身的恢复续跑。
|
||
var execUserId int64
|
||
if u, e := utils.GetUserInfo(ctx); e == nil && u != nil {
|
||
execUserId = int64(u.Id)
|
||
}
|
||
// 全新执行与复用旧 ID 新建两条路径共用同一插入逻辑，收敛为 createExec
|
||
createExec := func() (int64, error) {
|
||
execId, err := sessionDao.ExecWorkflowDao.Insert(ctx, &sessionDto.CreateWorkflowReq{
|
||
UserId: execUserId,
|
||
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) {
|
||
err = fmt.Errorf("创建执行记录返回空ID")
|
||
}
|
||
if err != nil {
|
||
glog.Errorf(ctx, "工作流执行记录创建失败: %v", err)
|
||
return 0, err
|
||
}
|
||
return execId, nil
|
||
}
|
||
// 复用失败记录重跑：仅当上次执行为失败状态（executeOrResume 传入的 lastExec.Status）时重置复用；
|
||
// 上次成功 / 无历史 → 一律新建执行记录（createExec）。
|
||
// FlowExecutionStatus 是 *int8 别名，Code() 返回包级指针，直接 == 是地址比较恒为 false，
|
||
// 需解引用按值比较，否则复用失败记录时不会重置为 Running、也不更新 RequestParams
|
||
if execId > 0 && status != nil && *status == *flow.FlowExecutionStatusFailed.Code() {
|
||
// 换参全新跑：先取将被废弃的旧逻辑运行(组)（ResetRunning 随即会把它覆盖成新组，须在重置前读）。
|
||
// 旧组此前属"失败可续跑"保留态，现用户改参数改走全新运行，旧组永不再续跑 → 重置成功后软删其
|
||
// checkpoint/段/异步残留回收（软删即终态，组不复活）。新组由下方 launchExecution 使用。
|
||
var oldGroup string
|
||
if prev, e := sessionDao.ExecWorkflowDao.GetById(ctx, execId); e == nil && prev != nil {
|
||
oldGroup = prev.NodeGroupId
|
||
}
|
||
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
|
||
}
|
||
// 已抢到重置权即本组唯一执行者，可安全回收旧组（无并发续跑方会读它）
|
||
if oldGroup != "" {
|
||
_ = flowDao.FlowCheckpointDao.Delete(ctx, oldGroup)
|
||
_ = flowDao.FlowAsyncTaskDao.DeleteByGroup(ctx, oldGroup)
|
||
_ = flowDao.FlowSegmentResultDao.DeleteByGroup(ctx, oldGroup)
|
||
}
|
||
} else {
|
||
execId, err = createExec()
|
||
if err != nil {
|
||
return
|
||
}
|
||
}
|
||
|
||
return launchExecution(ctx, conn, execId, req.FlowId, nodeGroupId, conn.SessionId, req.FlowContent, true)
|
||
}
|
||
|
||
// 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
|
||
}
|
||
// 续跑 = 同一逻辑运行的延续：复用 exec 记录的组（读其 checkpoint/段/异步活行继续），不换组——
|
||
// 换组会读不到上一 attempt 写在该组下的断点而从图头重跑。仅旧记录无组时新造并随重置持久化。
|
||
nodeGroupId := flowInfo.NodeGroupId
|
||
if nodeGroupId == "" {
|
||
nodeGroupId = uuid.NewString()
|
||
}
|
||
reset, err := sessionDao.ExecWorkflowDao.ResetRunning(ctx, flowInfo.Id, nodeGroupId, prevStatus)
|
||
if err != nil {
|
||
return
|
||
}
|
||
if !reset {
|
||
// 状态已被其它路径（恢复例程/并发触发）抢先重置为运行中：放弃续跑
|
||
return flowInfo.Id, errExecAlreadyRunning
|
||
}
|
||
return launchExecution(ctx, nil, flowInfo.Id, flowInfo.FlowId, nodeGroupId, flowInfo.SessionId, flowInfo.RequestParams, false)
|
||
}
|
||
|
||
// launchExecution execute 与 reExecute 共用的启动尾部：计费建单 →（可选）推送 round_start →
|
||
// 心跳 → hub 接管 → BuildExecution。
|
||
// conn 非 nil（WS 路径）时在计费通过后推送 round_start；reExecute 无连接传 nil 不推送。
|
||
// 返回语义与调用方原始约定一致：计费门禁失败返回 (execId, err)（终局失败），
|
||
// BuildExecution 失败返回 (execId, err)（execId 已创建/复用，wrapper 凭其落终态并结算），
|
||
// 成功返回 (execId, nil)。
|
||
func launchExecution(ctx context.Context, conn *wsCommon.WsConnection, execId int64, flowId int64, nodeGroupId string, sessionId string, flowContent *entity.FlowInfo, forceNewRun bool) (id int64, err error) {
|
||
// 工作流计费：建计费单（门禁：余额>=min_balance，钱包须存在）。
|
||
// 业务错误（余额不足/钱包不可用/费率非法）→ errBillingGateBlocked 终局失败，不重试不恢复；
|
||
// 续跑复用 execId，原计费单仍 CREATED 则幂等沿用、原单已终态则开新单
|
||
if err := openBillingOrder(ctx, execId, flowContent); err != nil {
|
||
return execId, err
|
||
}
|
||
if conn != nil {
|
||
_ = 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：确认真正启动执行前标记 owned，供并发附着连接识别运行中持有者
|
||
if h := getProgressHub(ctx); h != nil && !h.Owned() {
|
||
h.MarkOwned()
|
||
}
|
||
if err = BuildExecution(ctx, forceNewRun, flowId, execId, nodeGroupId, sessionId, flowContent); err != nil {
|
||
// 报错也返回真实 execId（与本函数计费门禁失败 return execId, err 一致）：execute/reExecute
|
||
// 已创建或复用了执行记录，wrapper（handleExecute）必须拿到它才能 recordWorkflow 落终态并结算。
|
||
// 原来丢 id 返回 (0, err) 会让 wrapper 退化为 recordExecutionFailure 兜底（按会话+流程查最近
|
||
// Running 记录）；当图内已提前触发过 summary/记录状态不再是 Running 时兜底会静默跳过，
|
||
// 结算永不执行 → 计费单遗留 CREATED（取消/中断漏扣，2026-09-03 已修，见 lambda_summary.go）。
|
||
return execId, err
|
||
}
|
||
return execId, 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
|
||
// =========================================================================
|
||
configMap := buildConfigMap(flowContent, nodeList)
|
||
|
||
// =========================================================================
|
||
// 构建全局执行入参
|
||
// =========================================================================
|
||
execInput := &flowDto.FlowExecutionInput{
|
||
NodeGroupId: nodeGroupId,
|
||
ExecutionId: executionId,
|
||
FlowId: flowId,
|
||
ConfigMap: configMap,
|
||
SessionId: sessionId,
|
||
ForceNewRun: forceNewRun,
|
||
}
|
||
|
||
// 全新执行/换参重跑(forceNewRun)无需起跑前清理：checkpoint/async/segment 均以 node_group_id
|
||
// (逻辑运行标识)为键，forceNewRun 换新组 = 全新键，天然不命中任何旧残留（也无软删墓碑可碰撞）；
|
||
// 被废弃的旧组残留由 execute() 复用失败 exec 时在重置成功后软删回收（见 execute）。
|
||
// 运行中/可续跑组永不删除。
|
||
// 驱动循环：编译期已对每个业务节点注册 WithInterruptAfterNodes（graph_build.go），节点正常完成后
|
||
// Eino 自动暂停并落 checkpoint。这里识别出"纯进度暂停"后同 checkpoint id 立即续跑，直至图完整跑完
|
||
// (err==nil) 或遇到真正终态（节点失败 / 用户取消 / 非中断错误）。崩溃硬杀恢复 BuildExecution(false)
|
||
// 走同一循环：查得断点即从断点续跑，已完成的同步节点不再重跑/重复计费。
|
||
// 判别器：节点失败中断 RerunNodes 恒非空（HandleFailedNodeExecution→compose.Interrupt）；ctx 取消由
|
||
// 下方 ctx.Err() 短路；异步模型调用是阻塞式（async.go），不产生空 RerunNodes 伪暂停。故 RerunNodes
|
||
// 为空 = 编译期纯进度暂停 → 续跑。详见《工作流节点断点续跑技术设计.md》。
|
||
// WithForceNewRun 只允许出现在全新跑首轮（此时起跑前三清也已只做一次）；续跑/暂停轮绝不能带，
|
||
// 否则把断点续跑打成"忽略断点从图头重跑"。
|
||
first := forceNewRun
|
||
maxIter := len(flowContent.Nodes)*2 + 8 // 纯进度暂停每轮必推进 ≥1 节点, 正常轮数 ≤ 节点数+1; 超限即疑似死循环
|
||
iter := 0
|
||
for {
|
||
runOpts := []compose.Option{compose.WithCheckPointID(nodeGroupId)}
|
||
if first {
|
||
runOpts = append(runOpts, compose.WithForceNewRun())
|
||
first = false
|
||
}
|
||
_, err = runGraph.Invoke(ctx, execInput, runOpts...)
|
||
if err == nil {
|
||
break // 图完整跑完 → 下方成功尾部三清
|
||
}
|
||
// 图执行被 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 {
|
||
return fmt.Errorf("执行工作流失败: %v", err)
|
||
}
|
||
if len(info.RerunNodes) == 0 {
|
||
// 纯进度暂停（编译期 after-node checkpoint 已落库）→ 同 id 续跑下一段
|
||
iter++
|
||
if iter > maxIter {
|
||
return fmt.Errorf("执行工作流失败: 断点续跑超限(%d 次), 疑似死循环", maxIter)
|
||
}
|
||
continue
|
||
}
|
||
// 节点失败中断 → 终态失败
|
||
var sb strings.Builder
|
||
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)
|
||
fmt.Fprintf(&sb, "\n节点：%v, 失败原因：%v", valMap["node"], valMap["error"])
|
||
}
|
||
}
|
||
if sb.Len() > 0 {
|
||
err = fmt.Errorf("%v个节点，%v", errNodeCount, strings.TrimPrefix(sb.String(), "\n"))
|
||
}
|
||
return fmt.Errorf("执行工作流失败: %v", err)
|
||
}
|
||
// 执行成功：软删本逻辑运行(组)的 checkpoint/段/异步缓存（终态组，此后不再被读写，软删不复活）。
|
||
// 失败/取消不走到这里，组行保留供 reExecute / 恢复续跑复用；换参重跑由 execute 回收旧组。
|
||
_ = flowDao.FlowCheckpointDao.Delete(ctx, nodeGroupId)
|
||
_ = flowDao.FlowSegmentResultDao.DeleteByGroup(ctx, nodeGroupId)
|
||
_ = flowDao.FlowAsyncTaskDao.DeleteByGroup(ctx, nodeGroupId)
|
||
return
|
||
}
|
||
|
||
// SessionWsService 会话 WebSocket 服务器：普通对话与工作流共用一条连接，
|
||
// 首次连接仅升级，后续按消息 type 路由到对话/工作流处理器
|
||
// （对话处理器在 react_ws_exec.go 注册，工作流处理器在上方 init 注册）。
|
||
var SessionWsService = wsCommon.NewWsServer(
|
||
wsCommon.WithConnKeyPrefix("ws:session:"),
|
||
)
|
||
|
||
// WsConnect 控制器统一入口：升级 WebSocket（普通对话/工作流均由消息 type 区分，此处不区分）
|
||
func WsConnect(ctx context.Context, r *ghttp.Request, req *sessionDto.WebSocketConnectReq) error {
|
||
_, err := SessionWsService.Upgrade(ctx, r, req.SessionId)
|
||
return err
|
||
}
|
||
|
||
// ensureSession 解析前端 sessionId 并确保会话存在：命中已存在会话则复用其 id，否则按 name 新建。
|
||
// 普通对话（react_ws_exec.go）与工作流（exec_ws.go）共用。
|
||
func ensureSession(ctx context.Context, sessionId string, name string) error {
|
||
exist, err := sessionDao.SessionDao.GetById(ctx, sessionId)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
if exist != nil {
|
||
return nil
|
||
}
|
||
if r := []rune(name); len(r) > 128 { // session_name VARCHAR(128)
|
||
name = string(r[:128])
|
||
}
|
||
_, err = sessionDao.SessionDao.Insert(ctx, &sessionDto.CreateSessionReq{SessionId: sessionId, SessionName: name})
|
||
return err
|
||
}
|