fix(workflow): 修复取消/中断漏扣费与终态提前落库问题
- launchExecution 失败时返回真实 execId，避免 wrapper 兜底失效 - summary 节点移除执行终态与计费写库，统一由 recordWorkflow 结算 - 更新数据库连接凭据
This commit is contained in:
+4
-4
@@ -7,8 +7,8 @@ database:
|
||||
- type: "pgsql"
|
||||
host: "192.168.0.83"
|
||||
port: "15432"
|
||||
user: "postgres"
|
||||
pass: "Bjang09@686^*^"
|
||||
user: "sql9f15b63fd203b36e"
|
||||
pass: "1ec94b1acdaf57b66030242d418fad5a"
|
||||
name: "digital-human"
|
||||
prefix: "digital_human_" # (可选)表名前缀
|
||||
role: "master" # (可选)数据库主从角色(master/slave)，默认为master。如果不使用应用主从机制请不配置或留空即可。
|
||||
@@ -28,8 +28,8 @@ database:
|
||||
- type: "pgsql"
|
||||
host: "192.168.0.83"
|
||||
port: "15432"
|
||||
user: "postgres"
|
||||
pass: "Q!P@z#M$1@686^*^.."
|
||||
user: "sql9f15b63fd203b36e"
|
||||
pass: "1ec94b1acdaf57b66030242d418fad5a"
|
||||
name: "black-deacon"
|
||||
prefix: "black_deacon_" # (可选)表名前缀
|
||||
role: "master"
|
||||
|
||||
@@ -359,7 +359,8 @@ func reExecute(ctx context.Context, execWorkflowId int64, prevStatus int8) (id i
|
||||
// 心跳 → hub 接管 → BuildExecution。
|
||||
// conn 非 nil（WS 路径）时在计费通过后推送 round_start；reExecute 无连接传 nil 不推送。
|
||||
// 返回语义与调用方原始约定一致：计费门禁失败返回 (execId, err)（终局失败），
|
||||
// BuildExecution 失败返回 (0, err)，成功返回 (execId, nil)。
|
||||
// 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 终局失败，不重试不恢复；
|
||||
@@ -380,7 +381,12 @@ func launchExecution(ctx context.Context, conn *wsCommon.WsConnection, execId in
|
||||
h.MarkOwned()
|
||||
}
|
||||
if err = BuildExecution(ctx, forceNewRun, flowId, execId, nodeGroupId, sessionId, flowContent); err != nil {
|
||||
return 0, err
|
||||
// 报错也返回真实 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
|
||||
}
|
||||
|
||||
@@ -1,14 +1,10 @@
|
||||
package flow
|
||||
|
||||
import (
|
||||
"ai-agent/workflow/consts/flow"
|
||||
"ai-agent/workflow/consts/public"
|
||||
nodeDao "ai-agent/workflow/dao/node"
|
||||
sessionDao "ai-agent/workflow/dao/session"
|
||||
flowDto "ai-agent/workflow/model/dto/flow"
|
||||
nodeDto "ai-agent/workflow/model/dto/node"
|
||||
sessionDto "ai-agent/workflow/model/dto/session"
|
||||
"ai-agent/workflow/model/entity"
|
||||
"context"
|
||||
"fmt"
|
||||
"strings"
|
||||
@@ -33,47 +29,29 @@ func SummaryLambda(ctx context.Context, input any) (any, error) {
|
||||
return nil, fmt.Errorf("汇总节点入参类型错误，实际是 %T", input)
|
||||
}
|
||||
|
||||
// 聚合所有已执行节点中需入库的文件结果（两层规则）
|
||||
// 汇总节点只做本职：聚合本次执行已产出、需入库的文件结果（两层规则）并落 exec_workflow_result。
|
||||
// 终态（exec_workflow.status / total_tokens / total_fee）与计费**禁止在图内写**——
|
||||
// summary 会随子流程/多末端/断点续跑在整体还没跑完时提前执行，此处若把 exec 置成 status=2
|
||||
// "成功"，会骗过 wrapper 的失败兜底（recordExecutionFailure 只标记 Running 记录）→ 取消/中断后
|
||||
// recordWorkflow/settleBilling 永不触发，计费单遗留 CREATED、已消耗 token 漏扣
|
||||
//（线上 per_token/per_second 取消不扣费根因，2026-09-03）。终态只允许 BuildExecution 返回后的
|
||||
// recordWorkflow 单点落库并结算（含 total_tokens/total_fee/actual_amount 回填），图内不再越权写 exec 行。
|
||||
summaryResult := collectSaveFileResults(ctx, execInput.Global)
|
||||
|
||||
// 把汇总结果存入当前节点的输出
|
||||
g.Log().Info(ctx, fmt.Sprintf("结果汇总完成，汇总数据：%+v", summaryResult))
|
||||
|
||||
err := gfdb.DB(ctx, public.DbNameBlackDeacon).Transaction(ctx, func(ctx context.Context, tx gdb.TX) error {
|
||||
res, _, err := nodeDao.NodeExecutionDao.ListByFlowExecutionId(ctx, &nodeDto.ListNodeExecutionByFlowReq{
|
||||
NodeGroupId: execInput.Global.NodeGroupId,
|
||||
}, entity.NodeExecutionCol.TokenInfo)
|
||||
if err != nil {
|
||||
if len(summaryResult) > 0 {
|
||||
err := gfdb.DB(ctx, public.DbNameBlackDeacon).Transaction(ctx, func(ctx context.Context, tx gdb.TX) error {
|
||||
_, err := sessionDao.ExecWorkflowResultDao.BatchInsert(ctx, summaryResult)
|
||||
return err
|
||||
}
|
||||
var totalTokens int
|
||||
var totalFee float64
|
||||
for _, item := range res {
|
||||
for _, itemToken := range item.TokenInfo {
|
||||
m := gconv.Map(itemToken)
|
||||
totalTokens += gconv.Int(m["total_tokens"])
|
||||
totalFee += gconv.Float64(m["total_fee"])
|
||||
}
|
||||
}
|
||||
_, err = sessionDao.ExecWorkflowDao.Update(ctx, &sessionDto.UpdateWorkflowReq{
|
||||
Id: execInput.Global.ExecutionId,
|
||||
Status: flow.FlowExecutionStatusSuccess.Code(),
|
||||
TotalTokens: totalTokens,
|
||||
TotalFee: totalFee,
|
||||
})
|
||||
if err != nil {
|
||||
return err
|
||||
return nil, err
|
||||
}
|
||||
if len(summaryResult) > 0 {
|
||||
_, err = sessionDao.ExecWorkflowResultDao.BatchInsert(ctx, summaryResult)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
return nil
|
||||
})
|
||||
}
|
||||
|
||||
return execInput, err
|
||||
return execInput, nil
|
||||
}
|
||||
|
||||
// collectSaveFileResults 按两层规则收集需入库的文件结果：
|
||||
|
||||
Reference in New Issue
Block a user