From 73f296731ca50d2d076e59aa2fda58c22d95743d Mon Sep 17 00:00:00 2001 From: qhd <1766646056@qq.com> Date: Fri, 4 Sep 2026 17:36:46 +0800 Subject: [PATCH] =?UTF-8?q?refactor(workflow):=20=E9=9A=94=E7=A6=BB?= =?UTF-8?q?=E9=80=BB=E8=BE=91=E8=BF=90=E8=A1=8C=E9=94=AE=E4=B8=BA=20node?= =?UTF-8?q?=5Fgroup=5Fid?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 将 checkpoint、异步任务与段结果缓存的唯一键从 execution_id 改为 node_group_id，区分换参重跑与续跑语义；恢复/续跑复用执行记录的组标识，换参重跑则换新组并软删旧组残留，消除软删墓碑导致同键重存失效的问题。 --- update.sql | 50 +++++++++++++++++ workflow/dao/flow/flow_async_task_dao.go | 49 ++++++++--------- workflow/dao/flow/flow_segment_result_dao.go | 38 +++++++------ workflow/model/entity/flow_async_task.go | 21 +++++--- workflow/model/entity/flow_segment_result.go | 7 ++- workflow/service/flow/async.go | 23 ++++---- workflow/service/flow/exec_recover.go | 7 ++- workflow/service/flow/exec_ws.go | 57 ++++++++++---------- workflow/service/flow/lambda_core.go | 8 +-- workflow/service/flow/model_call.go | 4 +- 10 files changed, 167 insertions(+), 97 deletions(-) diff --git a/update.sql b/update.sql index 57527f0..333ecd8 100644 --- a/update.sql +++ b/update.sql @@ -929,3 +929,53 @@ COMMENT ON COLUMN black_deacon_exec_workflow.user_id IS '执行用户ID（数字 --------------------终局回填业务实扣：exec_workflow.actual_amount（用户钱包实际扣除金额，元；区别于 total_fee=模型按次费用合计）--------------------- ALTER TABLE black_deacon_exec_workflow ADD COLUMN IF NOT EXISTS actual_amount NUMERIC(15,2) NOT NULL DEFAULT 0; -- 业务扣费（settle/cancel 结算实收；失败/未结算=0） COMMENT ON COLUMN black_deacon_exec_workflow.actual_amount IS '业务扣费（用户钱包实际扣除金额，元；结算/取消回填实收，失败/未结算=0，区别于 total_fee=模型按次费用合计）'; + +--------------------三表键按逻辑运行(node_group_id)隔离：exec_workflow 补列回填 + async/segment 加列换键 + checkpoint 活行归组（2026-09-04，详见《工作流状态表键按逻辑运行隔离技术设计.md》）-------------------- +-- 根因：flow_checkpoint 软删(execution_id 键)后同键 ON CONFLICT 重存不清 deleted_at → 续跑读不到断点从图头转圈。 +-- 方案：exec_workflow 持久化 node_group_id（逻辑运行标识：续跑复用、forceNewRun 换新 uuid）； +-- checkpoint_id / async / segment 唯一键一律纳入该组；同组重复只 Upsert 重置、软删即终态不复活、无物理删。 +-- 幂等：全程 ADD COLUMN IF NOT EXISTS / UPDATE 仅改空值 / DROP+CREATE UNIQUE IF EXISTS。 +-- 1) exec_workflow.node_group_id：仓库此前漏建该列 DDL（生产早期手工加列则此处 no-op）。 +-- 历史行回填自身 id(=其遗留 checkpoint/async/segment 旧键)：部署前中断/失败可续跑的 exec 无缝衔接到新组键续跑； +-- 已带真实 uuid 现役组的行（新代码产生的）不受影响。 +ALTER TABLE black_deacon_exec_workflow ADD COLUMN IF NOT EXISTS node_group_id VARCHAR(64) NOT NULL DEFAULT ''; +COMMENT ON COLUMN black_deacon_exec_workflow.node_group_id IS '节点组ID（逻辑运行标识：续跑复用，forceNewRun 换新 uuid）'; +UPDATE black_deacon_exec_workflow SET node_group_id = id::text WHERE node_group_id IS NULL OR node_group_id = ''; + +-- 2) async/segment 加列并按 exec 现役组回填。NULLIF 防空组并到 execution_id::text（=旧"exec 作用域"读集，等价不丢不并） +ALTER TABLE black_deacon_flow_async_task ADD COLUMN IF NOT EXISTS node_group_id VARCHAR(64); +ALTER TABLE black_deacon_flow_segment_result ADD COLUMN IF NOT EXISTS node_group_id VARCHAR(64); +UPDATE black_deacon_flow_async_task a SET node_group_id = COALESCE(NULLIF(e.node_group_id,''), a.execution_id::text) + FROM black_deacon_exec_workflow e WHERE a.node_group_id IS NULL AND e.id = a.execution_id; +UPDATE black_deacon_flow_segment_result s SET node_group_id = COALESCE(NULLIF(e.node_group_id,''), s.execution_id::text) + FROM black_deacon_exec_workflow e WHERE s.node_group_id IS NULL AND e.id = s.execution_id; +-- exec 行缺失的孤儿兜底（理论无）：按 execution_id 文本伪组，保证唯一性成立 +UPDATE black_deacon_flow_async_task SET node_group_id = execution_id::text WHERE node_group_id IS NULL; +UPDATE black_deacon_flow_segment_result SET node_group_id = execution_id::text WHERE node_group_id IS NULL; +ALTER TABLE black_deacon_flow_async_task ALTER COLUMN node_group_id SET NOT NULL; +ALTER TABLE black_deacon_flow_segment_result ALTER COLUMN node_group_id SET NOT NULL; + +-- 3) 换唯一键：旧 (execution_id,node_id,segment_index) → 新 (node_group_id,node_id,segment_index) +DROP INDEX IF EXISTS uk_async_task_exec_node_seg; +CREATE UNIQUE INDEX IF NOT EXISTS uk_async_task_group_node_seg ON black_deacon_flow_async_task(node_group_id, node_id, segment_index); +DROP INDEX IF EXISTS uk_segment_result_exec_node_idx; +CREATE UNIQUE INDEX IF NOT EXISTS uk_segment_result_group_node_idx ON black_deacon_flow_segment_result(node_group_id, node_id, segment_index); + +-- 4) checkpoint 表零 DDL：把仍活的断点从"checkpoint_id=executionId(数值串)"归一到"checkpoint_id=exec 现役组"， +-- 使新组键读到部署前中断/失败的断点。历史 exec 组=自身 id → 天然不变；仅 uuid 现役组发生重命名 +-- （单 exec 至多一行活断点；组为 uuid 不与遗留数值串冲突；守卫排除空串/自同值防并桶） +UPDATE black_deacon_flow_checkpoint c SET checkpoint_id = e.node_group_id + FROM black_deacon_exec_workflow e + WHERE c.deleted_at IS NULL AND c.checkpoint_id = e.id::text + AND e.node_group_id IS NOT NULL AND e.node_group_id <> '' AND e.node_group_id <> e.id::text; + +-- 5) 存量毒槽自愈：旧 forceNewRun 三清把"该 exec 仍可续跑"的 checkpoint 槽软删成了墓碑；uk_checkpoint_id 无条件唯一、 +-- 墓碑仍占槽，Save 的 ON CONFLICT 只会再 upsert 进墓碑且 deleted_at 不清 → 续跑永远读不到断点 = "从头转圈"现场。 +-- 软删即终态、不复活：被占槽不可再用，故给"可续跑(status 1/3)但当前槽无活断点"的 exec 重指全新 uuid 组， +-- 下次续跑写新活行、此后断点正常落库可断点续跑。代价：这批 exec 历史段/异步缓存(按旧 id 组)随之失联， +-- 属毒槽必然代价、只影响存量毒槽行；正常失败(槽上活断点在)的 exec 不动、断点续跑无缝衔接。 +UPDATE black_deacon_exec_workflow e + SET node_group_id = gen_random_uuid()::text + WHERE e.status IN (1, 3) + AND NOT EXISTS (SELECT 1 FROM black_deacon_flow_checkpoint c + WHERE c.deleted_at IS NULL AND c.checkpoint_id = e.node_group_id); diff --git a/workflow/dao/flow/flow_async_task_dao.go b/workflow/dao/flow/flow_async_task_dao.go index 106963e..de8b98c 100644 --- a/workflow/dao/flow/flow_async_task_dao.go +++ b/workflow/dao/flow/flow_async_task_dao.go @@ -4,7 +4,6 @@ import ( "ai-agent/workflow/consts/public" "ai-agent/workflow/model/entity" "context" - "fmt" "gitea.redpowerfuture.com/red-future/common/db/gfdb" ) @@ -20,10 +19,16 @@ var FlowAsyncTaskDao = &flowAsyncTaskDao{} type flowAsyncTaskDao struct{} -// Get 查询唯一键 (execution_id, node_id, segment_index) 的记录 -func (d *flowAsyncTaskDao) Get(ctx context.Context, execId int64, nodeId string, segIdx int) (res *entity.FlowAsyncTask, err error) { +// 缓存唯一键是 (node_group_id, node_id, segment_index)：node_group_id 是"逻辑运行(attempt)"标识， +// 全新/换参重跑换新组 = 换新键；续跑/恢复复用 exec 记录的组 = 读到同组活行。 +// 删除只有软删且只作用于"终态组"（成功尾部 / 换参重跑废弃的旧组），此类组此后永不再写， +// 软删行不会被同键重存（组永不复活）——与 flow_segment_result 同一约定。 +// execution_id 仅作溯源保留，不参与唯一键。 + +// Get 查询唯一键 (node_group_id, node_id, segment_index) 的记录 +func (d *flowAsyncTaskDao) Get(ctx context.Context, nodeGroupId string, nodeId string, segIdx int) (res *entity.FlowAsyncTask, err error) { r, err := gfdb.DB(ctx, public.DbNameBlackDeacon).Model(ctx, public.TableNameFlowAsyncTask). - Where(entity.FlowAsyncTaskCol.ExecutionId, execId). + Where(entity.FlowAsyncTaskCol.NodeGroupId, nodeGroupId). Where(entity.FlowAsyncTaskCol.NodeId, nodeId). Where(entity.FlowAsyncTaskCol.SegmentIndex, segIdx). One() @@ -43,8 +48,12 @@ func (d *flowAsyncTaskDao) Get(ctx context.Context, execId int64, nodeId string, // 提交时本无结果，统一写 '{}' 占位（与表列默认值一致）。 // 2. 必须用 OnDuplicate 限定冲突更新列——GoFrame Save 默认把 Data 里所有列写进 // ON CONFLICT DO UPDATE SET，若不限定会把已缓存的结果覆盖成 '{}'，与"保留已完成结果"矛盾。 -func (d *flowAsyncTaskDao) Upsert(ctx context.Context, execId int64, nodeId string, segIdx int, modelId, taskId int64, msgTopic string) error { +// +// 提交/重提统一走本函数重置同组活行（state→in-flight + 新 task_id/msg_topic）：不再"删行重建"， +// 活行从未被软删，OnConflict 更新即可复位，无墓碑冲突、无物理删除。 +func (d *flowAsyncTaskDao) Upsert(ctx context.Context, nodeGroupId string, execId int64, nodeId string, segIdx int, modelId, taskId int64, msgTopic string) error { rec := &entity.FlowAsyncTask{ + NodeGroupId: nodeGroupId, ExecutionId: execId, NodeId: nodeId, SegmentIndex: segIdx, @@ -57,7 +66,7 @@ func (d *flowAsyncTaskDao) Upsert(ctx context.Context, execId int64, nodeId stri _, err := gfdb.DB(ctx, public.DbNameBlackDeacon). Model(ctx, public.TableNameFlowAsyncTask). Data(rec). - OnConflict(entity.FlowAsyncTaskCol.ExecutionId, entity.FlowAsyncTaskCol.NodeId, entity.FlowAsyncTaskCol.SegmentIndex). + OnConflict(entity.FlowAsyncTaskCol.NodeGroupId, entity.FlowAsyncTaskCol.NodeId, entity.FlowAsyncTaskCol.SegmentIndex). OnDuplicate(entity.FlowAsyncTaskCol.ModelId, entity.FlowAsyncTaskCol.TaskId, entity.FlowAsyncTaskCol.MsgTopic, entity.FlowAsyncTaskCol.State). Save() return err @@ -65,13 +74,13 @@ func (d *flowAsyncTaskDao) Upsert(ctx context.Context, execId int64, nodeId stri // UpdateByKey 按唯一键更新 state/result（OmitNil 丢弃 nil 字段，map 值非 nil 全写入； // state=0 也能落库）。Result 列是 JSONB，空串无法写入，统一落 '{}' 表示无结果。 -func (d *flowAsyncTaskDao) UpdateByKey(ctx context.Context, execId int64, nodeId string, segIdx int, state int, result string) error { +func (d *flowAsyncTaskDao) UpdateByKey(ctx context.Context, nodeGroupId string, nodeId string, segIdx int, state int, result string) error { resultVal := result if resultVal == "" { resultVal = "{}" } _, err := gfdb.DB(ctx, public.DbNameBlackDeacon).Model(ctx, public.TableNameFlowAsyncTask). - Where(entity.FlowAsyncTaskCol.ExecutionId, execId). + Where(entity.FlowAsyncTaskCol.NodeGroupId, nodeGroupId). Where(entity.FlowAsyncTaskCol.NodeId, nodeId). Where(entity.FlowAsyncTaskCol.SegmentIndex, segIdx). Data(map[string]any{ @@ -82,22 +91,14 @@ func (d *flowAsyncTaskDao) UpdateByKey(ctx context.Context, execId int64, nodeId return err } -// DeleteByKey 物理删除：实体嵌 SQLBaseDO 软删后同键重存无法复活（ON CONFLICT 不含 deleted_at）， -// 与 flow_segment_result 相同约束，须 raw Exec 用物理全名 -func (d *flowAsyncTaskDao) DeleteByKey(ctx context.Context, execId int64, nodeId string, segIdx int) error { - const physicalTable = "black_deacon_flow_async_task" +// DeleteByGroup 软删指定逻辑运行(组)的异步任务缓存（gfdb Model.Delete 在 deletedAt 配置下退化为软删）。 +// 仅允许对"终态组"调用：① 工作流执行成功后（BuildExecution 尾部）；② 同一条 exec 换参重跑（forceNewRun） +// 废弃的旧组（execute 重置成功后回收）。终态组此后永不再被读写 → 软删行不复活、不占同键写入冲突。 +// 失败/取消/重试耗尽一律不删（保留组行供同参数续跑复用）。绝无物理删除。 +func (d *flowAsyncTaskDao) DeleteByGroup(ctx context.Context, nodeGroupId string) error { _, err := gfdb.DB(ctx, public.DbNameBlackDeacon). - Exec(ctx, fmt.Sprintf("DELETE FROM %s WHERE execution_id = ? AND node_id = ? AND segment_index = ?", physicalTable), execId, nodeId, segIdx) - return err -} - -// DeleteByExecution 清理指定执行的异步任务缓存。统一清理策略（Task 11 方案A，无周期兜底）： -// 仅在两处 exec 级删除点调用——① 工作流执行成功后（BuildExecution 尾部，与 checkpoint/段清理同处）； -// ② 同一条 exec 以"全新跑"重开（forceNewRun 起跑前，参数已变，旧异步结果必须作废防误复用）。 -// 失败/取消/重试耗尽一律保留产物，供同参数手动续跑（reExecute）复用；不作节点级或周期清扫。 -func (d *flowAsyncTaskDao) DeleteByExecution(ctx context.Context, execId int64) error { - const physicalTable = "black_deacon_flow_async_task" - _, err := gfdb.DB(ctx, public.DbNameBlackDeacon). - Exec(ctx, fmt.Sprintf("DELETE FROM %s WHERE execution_id = ?", physicalTable), execId) + Model(ctx, public.TableNameFlowAsyncTask). + Where(entity.FlowAsyncTaskCol.NodeGroupId, nodeGroupId). + Delete() return err } diff --git a/workflow/dao/flow/flow_segment_result_dao.go b/workflow/dao/flow/flow_segment_result_dao.go index b072c9c..e6b6e8b 100644 --- a/workflow/dao/flow/flow_segment_result_dao.go +++ b/workflow/dao/flow/flow_segment_result_dao.go @@ -4,7 +4,6 @@ import ( "ai-agent/workflow/consts/public" "ai-agent/workflow/model/entity" "context" - "fmt" "gitea.redpowerfuture.com/red-future/common/db/gfdb" ) @@ -13,9 +12,16 @@ var FlowSegmentResultDao = &flowSegmentResultDao{} type flowSegmentResultDao struct{} -// Save 段成功后落库：唯一键 (execution_id, node_id, segment_index)，冲突则更新 -func (d *flowSegmentResultDao) Save(ctx context.Context, execId int64, nodeId string, segmentIndex int, videoKey, videoURL string) error { +// 缓存唯一键是 (node_group_id, node_id, segment_index)：node_group_id 是"逻辑运行(attempt)"标识， +// 全新/换参重跑换新组 = 换新键；续跑/恢复复用 exec 记录的组 = 读到同组活行。 +// 删除只有软删且只作用于"终态组"（成功尾部 / 换参重跑废弃的旧组），此类组此后永不再写 → +// 软删行永不被同键重存（不复活）。实体嵌入 SQLBaseDO(含 deleted_at)，gfdb Model.Delete() 即软删， +// 不再需要 raw Exec 物理删除。 + +// Save 段成功后落库：唯一键 (node_group_id, node_id, segment_index)，冲突则更新视频引用 +func (d *flowSegmentResultDao) Save(ctx context.Context, nodeGroupId string, execId int64, nodeId string, segmentIndex int, videoKey, videoURL string) error { rec := &entity.FlowSegmentResult{ + NodeGroupId: nodeGroupId, ExecutionId: execId, NodeId: nodeId, SegmentIndex: segmentIndex, @@ -25,17 +31,17 @@ func (d *flowSegmentResultDao) Save(ctx context.Context, execId int64, nodeId st _, err := gfdb.DB(ctx, public.DbNameBlackDeacon). Model(ctx, public.TableNameFlowSegmentResult). Data(rec). - OnConflict(entity.FlowSegmentResultCol.ExecutionId, entity.FlowSegmentResultCol.NodeId, entity.FlowSegmentResultCol.SegmentIndex). + OnConflict(entity.FlowSegmentResultCol.NodeGroupId, entity.FlowSegmentResultCol.NodeId, entity.FlowSegmentResultCol.SegmentIndex). Save() return err } -// ListByNode 返回该节点已成功段（段序号 → 视频引用） -func (d *flowSegmentResultDao) ListByNode(ctx context.Context, execId int64, nodeId string) (map[int]entity.SegmentRef, error) { +// ListByNode 返回该节点已成功段（段序号 → 视频引用）；仅读当前逻辑运行(组)的活行 +func (d *flowSegmentResultDao) ListByNode(ctx context.Context, nodeGroupId string, nodeId string) (map[int]entity.SegmentRef, error) { var list []*entity.FlowSegmentResult err := gfdb.DB(ctx, public.DbNameBlackDeacon). Model(ctx, public.TableNameFlowSegmentResult). - Where(entity.FlowSegmentResultCol.ExecutionId, execId). + Where(entity.FlowSegmentResultCol.NodeGroupId, nodeGroupId). Where(entity.FlowSegmentResultCol.NodeId, nodeId). Scan(&list) if err != nil { @@ -48,16 +54,14 @@ func (d *flowSegmentResultDao) ListByNode(ctx context.Context, execId int64, nod return m, nil } -// DeleteByExecution 清理指定执行的段结果（全新执行前清旧段 / 工作流执行成功后清理）。 -// 必须物理删除：实体嵌入 SQLBaseDO(含 deleted_at) 会让 gfdb Model.Delete() 退化为软删除， -// 而 Save 的 ON CONFLICT DO UPDATE SET 不含 deleted_at（OmitNil 丢弃 nil）， -// 软删后同键重存无法复活该行 → 续跑复用静默失效（每次重执行都全量重生成）。 -func (d *flowSegmentResultDao) DeleteByExecution(ctx context.Context, execId int64) error { - // 表名须用物理全名：raw Exec 不经过 GoFrame 的 config prefix（black_deacon_）自动加前缀， - // 与 update.sql 物理建表名 black_deacon_flow_segment_result 保持一致；常量是短名 - // flow_segment_result（经 Model() 自动加前缀），不能在此复用。 - const physicalTable = "black_deacon_flow_segment_result" +// DeleteByGroup 软删指定逻辑运行(组)的段结果（gfdb Model.Delete 在 deletedAt 配置下退化为软删）。 +// 仅允许对"终态组"调用：① 工作流执行成功后（BuildExecution 尾部）；② 同一条 exec 换参重跑（forceNewRun） +// 废弃的旧组（execute 重置成功后回收）。终态组此后永不再被读写 → 软删行不复活。 +// 失败/取消不删（保留组行供 reExecute 复用已成功段）。绝无物理删除。 +func (d *flowSegmentResultDao) DeleteByGroup(ctx context.Context, nodeGroupId string) error { _, err := gfdb.DB(ctx, public.DbNameBlackDeacon). - Exec(ctx, fmt.Sprintf("DELETE FROM %s WHERE execution_id = ?", physicalTable), execId) + Model(ctx, public.TableNameFlowSegmentResult). + Where(entity.FlowSegmentResultCol.NodeGroupId, nodeGroupId). + Delete() return err } diff --git a/workflow/model/entity/flow_async_task.go b/workflow/model/entity/flow_async_task.go index 0951781..0ac8473 100644 --- a/workflow/model/entity/flow_async_task.go +++ b/workflow/model/entity/flow_async_task.go @@ -6,18 +6,22 @@ import "gitea.redpowerfuture.com/red-future/common/beans" // 崩溃恢复靠持久化的 msg_topic 重订阅 NATS 拿回结果，避免重复调用模型 type FlowAsyncTask struct { beans.SQLBaseDO `orm:",inherit"` - ExecutionId int64 `orm:"execution_id" json:"executionId" description:"所属执行ID"` - NodeId string `orm:"node_id" json:"nodeId" description:"所属节点ID"` - SegmentIndex int `orm:"segment_index" json:"segmentIndex" description:"段序号；非段调用为-1"` - ModelId int64 `orm:"model_id" json:"modelId" description:"模型ID"` - TaskId int64 `orm:"task_id" json:"taskId" description:"model-gateway任务ID"` - MsgTopic string `orm:"msg_topic" json:"msgTopic" description:"结果消息主题"` - State int `orm:"state" json:"state" description:"0=in-flight,1=done,2=failed"` - Result string `orm:"result" json:"result" description:"成功结果JSON(ModelCallRes)"` + // NodeGroupId 所属逻辑运行(attempt)标识：全新/换参重跑=新组，续跑/恢复/自动重试=复用 exec 记录的组。 + // 缓存唯一键以 (node_group_id, node_id, segment_index) 隔离，杜绝换参重跑软删墓碑与重跑同键写入冲突。 + NodeGroupId string `orm:"node_group_id" json:"nodeGroupId" description:"所属逻辑运行(node_group_id)标识"` + ExecutionId int64 `orm:"execution_id" json:"executionId" description:"所属执行ID（溯源，不参与唯一键）"` + NodeId string `orm:"node_id" json:"nodeId" description:"所属节点ID"` + SegmentIndex int `orm:"segment_index" json:"segmentIndex" description:"段序号；非段调用为-1"` + ModelId int64 `orm:"model_id" json:"modelId" description:"模型ID"` + TaskId int64 `orm:"task_id" json:"taskId" description:"model-gateway任务ID"` + MsgTopic string `orm:"msg_topic" json:"msgTopic" description:"结果消息主题"` + State int `orm:"state" json:"state" description:"0=in-flight,1=done,2=failed"` + Result string `orm:"result" json:"result" description:"成功结果JSON(ModelCallRes)"` } type flowAsyncTaskCol struct { beans.SQLBaseCol + NodeGroupId string ExecutionId string NodeId string SegmentIndex string @@ -30,6 +34,7 @@ type flowAsyncTaskCol struct { var FlowAsyncTaskCol = flowAsyncTaskCol{ SQLBaseCol: beans.DefSQLBaseCol, + NodeGroupId: "node_group_id", ExecutionId: "execution_id", NodeId: "node_id", SegmentIndex: "segment_index", diff --git a/workflow/model/entity/flow_segment_result.go b/workflow/model/entity/flow_segment_result.go index 490e132..6118ab6 100644 --- a/workflow/model/entity/flow_segment_result.go +++ b/workflow/model/entity/flow_segment_result.go @@ -6,7 +6,10 @@ import "gitea.redpowerfuture.com/red-future/common/beans" type FlowSegmentResult struct { beans.SQLBaseDO `orm:",inherit"` // 嵌入基础字段：Id, TenantId, Creator, CreatedAt, Updater, UpdatedAt, DeletedAt // 业务字段 - ExecutionId int64 `orm:"execution_id" json:"executionId" description:"执行ID"` + // NodeGroupId 所属逻辑运行(attempt)标识：段结果唯一键 (node_group_id, node_id, segment_index)， + // 换参重跑换组即换键，绝不复用被软删墓碑的旧组行（软删即终态，不复活） + NodeGroupId string `orm:"node_group_id" json:"nodeGroupId" description:"所属逻辑运行(node_group_id)标识"` + ExecutionId int64 `orm:"execution_id" json:"executionId" description:"执行ID（溯源，不参与唯一键）"` NodeId string `orm:"node_id" json:"nodeId" description:"视频生成节点ID"` SegmentIndex int `orm:"segment_index" json:"segmentIndex" description:"段序号"` VideoKey string `orm:"video_key" json:"videoKey" description:"视频输出字段key"` @@ -21,6 +24,7 @@ type SegmentRef struct { type flowSegmentResultCol struct { beans.SQLBaseCol + NodeGroupId string ExecutionId string NodeId string SegmentIndex string @@ -30,6 +34,7 @@ type flowSegmentResultCol struct { var FlowSegmentResultCol = flowSegmentResultCol{ SQLBaseCol: beans.DefSQLBaseCol, + NodeGroupId: "node_group_id", ExecutionId: "execution_id", NodeId: "node_id", SegmentIndex: "segment_index", diff --git a/workflow/service/flow/async.go b/workflow/service/flow/async.go index bc99e8c..293b672 100644 --- a/workflow/service/flow/async.go +++ b/workflow/service/flow/async.go @@ -91,12 +91,14 @@ func asyncCallAction(rec *entity.FlowAsyncTask) asyncAction { // 同步模型直接走 gateway.ModelCallResult，不落库（无恢复语义）。 // 注意：本函数是节点内阻塞调用（WaitModelCallResult 等回调），不是独立并发触发方， // 不参与 exec 并发仲裁（谁抢到执行权谁跑）——仲裁语义见《工作流执行并发仲裁设计.md》。 -func AsyncModelCallWithRecovery(ctx context.Context, execId int64, nodeId string, segIdx int, modelId int64, responseType model.ResponseType, sessionId string, requestParams map[string]any, businessParams map[string]any) (*gateway.ModelCallRes, error) { +// nodeGroupId 是逻辑运行(attempt)标识：缓存唯一键 (node_group_id,node_id,segment_index) 由它隔离， +// 续跑/恢复复用同组即可命中上一 attempt 的 in-flight/done 行（节点级重提/复用按 asyncCallAction）。 +func AsyncModelCallWithRecovery(ctx context.Context, nodeGroupId string, execId int64, nodeId string, segIdx int, modelId int64, responseType model.ResponseType, sessionId string, requestParams map[string]any, businessParams map[string]any) (*gateway.ModelCallRes, error) { if responseType == nil || *responseType != *model.ResponseTypeAsync.Code() { return gateway.ModelCallResult(ctx, modelId, responseType, sessionId, requestParams, businessParams) } - rec, err := flowDao.FlowAsyncTaskDao.Get(ctx, execId, nodeId, segIdx) + rec, err := flowDao.FlowAsyncTaskDao.Get(ctx, nodeGroupId, nodeId, segIdx) if err != nil { return nil, err } @@ -105,24 +107,21 @@ func AsyncModelCallWithRecovery(ctx context.Context, execId int64, nodeId string return unmarshalModelCallRes(rec.Result) case asyncActionFinalize: // 重订阅 msgTopic 收尾：拿回已发布结果（消息在 JetStream 保留 7 天）； - // 成功 → 落 done 复用；超时/订阅失败 → 清记录重提 + // 成功 → 落 done 复用；超时/订阅失败 → 落入下方重提（Upsert 重置同组活行，不删行） waitCtx, cancel := context.WithTimeout(ctx, asyncRecoverWaitTimeout) res, waitErr := gateway.WaitModelCallResult(waitCtx, rec.MsgTopic) cancel() if waitErr != nil { if errors.Is(waitErr, context.DeadlineExceeded) || errors.Is(waitErr, context.Canceled) { - _ = flowDao.FlowAsyncTaskDao.DeleteByKey(ctx, execId, nodeId, segIdx) break // 落入下方重新提交 } return nil, waitErr } - _ = flowDao.FlowAsyncTaskDao.UpdateByKey(ctx, execId, nodeId, segIdx, flowDao.FlowAsyncStateDone, marshalModelCallRes(res)) + _ = flowDao.FlowAsyncTaskDao.UpdateByKey(ctx, nodeGroupId, nodeId, segIdx, flowDao.FlowAsyncStateDone, marshalModelCallRes(res)) return res, nil case asyncActionResubmit: - // 清残留（failed 或 done 空结果） - if rec != nil { - _ = flowDao.FlowAsyncTaskDao.DeleteByKey(ctx, execId, nodeId, segIdx) - } + // 清残留语义改为下方 Upsert 重置：failed 或 done 空结果的行是活行（同组从未被软删）， + // OnConflict 直接复位为 in-flight + 新 task_id/msg_topic，无需也不可删行重建 } // 重新提交：先落库 in-flight（task_id/msg_topic），等待结果 @@ -130,16 +129,16 @@ func AsyncModelCallWithRecovery(ctx context.Context, execId int64, nodeId string if err != nil { return nil, err } - if err := flowDao.FlowAsyncTaskDao.Upsert(ctx, execId, nodeId, segIdx, modelId, res.TaskId, msgTopic); err != nil { + if err := flowDao.FlowAsyncTaskDao.Upsert(ctx, nodeGroupId, execId, nodeId, segIdx, modelId, res.TaskId, msgTopic); err != nil { return nil, err } waitRes, waitErr := gateway.WaitModelCallResult(ctx, msgTopic) if waitErr != nil { // 结果失败（model error/取消）：落 failed，调用方（段重试/恢复）决定后续 - _ = flowDao.FlowAsyncTaskDao.UpdateByKey(ctx, execId, nodeId, segIdx, flowDao.FlowAsyncStateFailed, "") + _ = flowDao.FlowAsyncTaskDao.UpdateByKey(ctx, nodeGroupId, nodeId, segIdx, flowDao.FlowAsyncStateFailed, "") return nil, waitErr } - _ = flowDao.FlowAsyncTaskDao.UpdateByKey(ctx, execId, nodeId, segIdx, flowDao.FlowAsyncStateDone, marshalModelCallRes(waitRes)) + _ = flowDao.FlowAsyncTaskDao.UpdateByKey(ctx, nodeGroupId, nodeId, segIdx, flowDao.FlowAsyncStateDone, marshalModelCallRes(waitRes)) return waitRes, nil } diff --git a/workflow/service/flow/exec_recover.go b/workflow/service/flow/exec_recover.go index d950f53..62d51ca 100644 --- a/workflow/service/flow/exec_recover.go +++ b/workflow/service/flow/exec_recover.go @@ -137,7 +137,12 @@ func recoverExecution(parentCtx context.Context, execId int64, attach *execAttac // 余额门禁（须 user.Id>0）——用 exec 所属租户合成系统用户 ctx（Id 取创建时落库的 user_id， // creator 仅 userName 推不回数字 id；旧记录 user_id=0 时其恢复续跑会被该门禁拦截），保留 span。 userCtx := context.WithValue(saveCtx, "user", &beans.User{Id: uint64(exec.UserId), UserName: exec.Creator, TenantId: exec.TenantId}) - nodeGroupId := uuid.NewString() // 置运行中与图执行的节点组标识须一致 + // 恢复 = 同一逻辑运行继续：复用 exec 记录的组读其 checkpoint 续跑（不换组，否则读不到断点从头跑）。 + // 仅旧记录无组时新造并随重置持久化；置运行中与图执行的组标识须一致。 + nodeGroupId := exec.NodeGroupId + if nodeGroupId == "" { + nodeGroupId = uuid.NewString() + } // 条件重置（原子防与用户断点续跑双跑）：仅当仍可恢复（status=3 或 status=1 心跳陈旧）时才抢到重置权 staleBeforeMs := time.Now().UnixMilli() - int64(heartbeatStaleAfter/time.Millisecond) reset, err := sessionDao.ExecWorkflowDao.ResetRunningIfRecoverable(userCtx, execId, nodeGroupId, staleBeforeMs) diff --git a/workflow/service/flow/exec_ws.go b/workflow/service/flow/exec_ws.go index 8717636..a5f58ca 100644 --- a/workflow/service/flow/exec_ws.go +++ b/workflow/service/flow/exec_ws.go @@ -306,6 +306,13 @@ func execute(ctx context.Context, conn *wsCommon.WsConnection, execId int64, sta // 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 { @@ -320,6 +327,12 @@ func execute(ctx context.Context, conn *wsCommon.WsConnection, execId int64, sta 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 { @@ -343,7 +356,12 @@ func reExecute(ctx context.Context, execWorkflowId int64, prevStatus int8) (id i if err != nil { return } - var nodeGroupId = uuid.NewString() + // 续跑 = 同一逻辑运行的延续：复用 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 @@ -419,26 +437,10 @@ func BuildExecution(ctx context.Context, forceNewRun bool, flowId, executionId i ForceNewRun: forceNewRun, } - // 全新执行前统一清理该执行残留（checkpoint + 段结果 + 异步任务缓存）： - // forceNewRun 复用同一条 exec 记录时可能留有旧参数/旧运行的产物，若不清： - // - 旧 checkpoint 会让"本应全新跑"误断点续跑（WithForceNewRun 虽绕开 Eino 读， - // 但 DB 残留键会被后续复用拾取，且成功尾部 Delete 也是清此键）； - // - 旧异步 done 结果会在 asyncCallAction 里被复用（key 含 execId+nodeId+segIdx， - // 参数已变时语义过期）； - // - 旧段结果被 lambda 复用生成错参视频。 - // 统一收口：只按 execution_id 清理（放在图启动前，避免多视频节点互相误删）。失败/取消不清， - // 交给成功尾部（下方同三处）或下一次 forceNewRun 起跑前统一清。 - if forceNewRun { - if err := flowDao.FlowCheckpointDao.Delete(ctx, gconv.String(executionId)); err != nil { - return fmt.Errorf("清理断点失败: %v", err) - } - if err := flowDao.FlowAsyncTaskDao.DeleteByExecution(ctx, executionId); err != nil { - return fmt.Errorf("清理异步任务缓存失败: %v", err) - } - if err := flowDao.FlowSegmentResultDao.DeleteByExecution(ctx, executionId); err != nil { - return fmt.Errorf("清理段结果失败: %v", err) - } - } + // 全新执行/换参重跑(forceNewRun)无需起跑前清理：checkpoint/async/segment 均以 node_group_id + // (逻辑运行标识)为键，forceNewRun 换新组 = 全新键，天然不命中任何旧残留（也无软删墓碑可碰撞）； + // 被废弃的旧组残留由 execute() 复用失败 exec 时在重置成功后软删回收（见 execute）。 + // 运行中/可续跑组永不删除。 // 驱动循环：编译期已对每个业务节点注册 WithInterruptAfterNodes（graph_build.go），节点正常完成后 // Eino 自动暂停并落 checkpoint。这里识别出"纯进度暂停"后同 checkpoint id 立即续跑，直至图完整跑完 // (err==nil) 或遇到真正终态（节点失败 / 用户取消 / 非中断错误）。崩溃硬杀恢复 BuildExecution(false) @@ -452,7 +454,7 @@ func BuildExecution(ctx context.Context, forceNewRun bool, flowId, executionId i maxIter := len(flowContent.Nodes)*2 + 8 // 纯进度暂停每轮必推进 ≥1 节点, 正常轮数 ≤ 节点数+1; 超限即疑似死循环 iter := 0 for { - runOpts := []compose.Option{compose.WithCheckPointID(gconv.String(executionId))} + runOpts := []compose.Option{compose.WithCheckPointID(nodeGroupId)} if first { runOpts = append(runOpts, compose.WithForceNewRun()) first = false @@ -497,12 +499,11 @@ func BuildExecution(ctx context.Context, forceNewRun bool, flowId, executionId i } return fmt.Errorf("执行工作流失败: %v", err) } - // 清理断点数据 - _ = flowDao.FlowCheckpointDao.Delete(ctx, gconv.String(executionId)) - // 清理该执行段结果，下次执行无残留（失败则保留，供 reExecute 复用） - _ = flowDao.FlowSegmentResultDao.DeleteByExecution(ctx, executionId) - // 清理该执行异步任务缓存，下次执行无残留（失败则保留，供恢复复用） - _ = flowDao.FlowAsyncTaskDao.DeleteByExecution(ctx, executionId) + // 执行成功：软删本逻辑运行(组)的 checkpoint/段/异步缓存（终态组，此后不再被读写，软删不复活）。 + // 失败/取消不走到这里，组行保留供 reExecute / 恢复续跑复用；换参重跑由 execute 回收旧组。 + _ = flowDao.FlowCheckpointDao.Delete(ctx, nodeGroupId) + _ = flowDao.FlowSegmentResultDao.DeleteByGroup(ctx, nodeGroupId) + _ = flowDao.FlowAsyncTaskDao.DeleteByGroup(ctx, nodeGroupId) return } diff --git a/workflow/service/flow/lambda_core.go b/workflow/service/flow/lambda_core.go index 0e37027..c13cc64 100644 --- a/workflow/service/flow/lambda_core.go +++ b/workflow/service/flow/lambda_core.go @@ -92,7 +92,7 @@ func ModelLambda(ctx context.Context, input any) (any, error) { // 续跑（!ForceNewRun）时读取该节点已成功段；全新执行不查（BuildExecution 已清旧段），saved 为 nil → 全量重生成 var saved map[int]entity.SegmentRef if !nodeInput.Global.ForceNewRun && segVideo { - saved, err = flowDao.FlowSegmentResultDao.ListByNode(ctx, nodeInput.Global.ExecutionId, cNodeId) + saved, err = flowDao.FlowSegmentResultDao.ListByNode(ctx, nodeInput.Global.NodeGroupId, cNodeId) if err != nil { return nil, err } @@ -115,7 +115,7 @@ func ModelLambda(ctx context.Context, input any) (any, error) { defer wg.Done() // 每段单次调用，不原地重试：段失败即走节点失败收口（HandleFailedNodeExecution → Interrupt）， // 下次 reExecute 由 planSegmentResume 复用已成功段、仅重生成失败段 - results[i], tokenRes[i], isInference[i], errs[i] = ModelCallResultLambda(ctx, nodeInput.Config.ModelConfig.ModelId, nodeInput.Global.SessionId, params, nodeInput.Config.Prompt, nodeInput.Global.ExecutionId, cNodeId, idxList[i]) + results[i], tokenRes[i], isInference[i], errs[i] = ModelCallResultLambda(ctx, nodeInput.Global.NodeGroupId, nodeInput.Config.ModelConfig.ModelId, nodeInput.Global.SessionId, params, nodeInput.Config.Prompt, nodeInput.Global.ExecutionId, cNodeId, idxList[i]) // 每段成功立即落库：该段刚成功即持久化，其他段仍在跑时已成功段也不丢； // 后续段失败或进程崩溃（panic/OOM/kill）时，已完成段已在库中，reExecute 可直接复用 if segVideo && errs[i] == nil { @@ -125,7 +125,7 @@ func ModelLambda(ctx context.Context, input any) (any, error) { if key == "" || url == "" { continue } - if err := flowDao.FlowSegmentResultDao.Save(ctx, nodeInput.Global.ExecutionId, cNodeId, idxList[i], key, url); err != nil { + if err := flowDao.FlowSegmentResultDao.Save(ctx, nodeInput.Global.NodeGroupId, nodeInput.Global.ExecutionId, cNodeId, idxList[i], key, url); err != nil { saveErrs[i] = err } } @@ -173,7 +173,7 @@ func ModelLambda(ctx context.Context, input any) (any, error) { } } else { for _, params := range paramsList { - res, modelRes, _, err := ModelCallResultLambda(ctx, nodeInput.Config.ModelConfig.ModelId, nodeInput.Global.SessionId, params, nodeInput.Config.Prompt, nodeInput.Global.ExecutionId, cNodeId, flowDao.FlowAsyncSegSentinel) + res, modelRes, _, err := ModelCallResultLambda(ctx, nodeInput.Global.NodeGroupId, nodeInput.Config.ModelConfig.ModelId, nodeInput.Global.SessionId, params, nodeInput.Config.Prompt, nodeInput.Global.ExecutionId, cNodeId, flowDao.FlowAsyncSegSentinel) if err != nil { return nil, err } diff --git a/workflow/service/flow/model_call.go b/workflow/service/flow/model_call.go index 27e756e..0b582bb 100644 --- a/workflow/service/flow/model_call.go +++ b/workflow/service/flow/model_call.go @@ -22,7 +22,7 @@ import ( // ModelCallResultLambda 调用模型并返回输出内容列表，同时回传本次调用的 token/费用（*gateway.ModelCallRes） // 与是否推理模型（供 ModelLambda 决定分批结果是否拼接），供调用方（ModelLambda）累计写入节点执行记录 // token_info，最后由汇总节点聚合到 exec_workflow。 -func ModelCallResultLambda(ctx context.Context, modelId int64, sessionId string, modelRequestParams map[string]any, prompt string, execId int64, nodeId string, segIdx int) ([]map[string]any, *gateway.ModelCallRes, bool, error) { +func ModelCallResultLambda(ctx context.Context, nodeGroupId string, modelId int64, sessionId string, modelRequestParams map[string]any, prompt string, execId int64, nodeId string, segIdx int) ([]map[string]any, *gateway.ModelCallRes, bool, error) { modelInfo, err := gateway.GetModelInfoById(ctx, &gateway.GetModelInfoByIdReq{ModelId: modelId}) if err != nil { return nil, nil, false, fmt.Errorf("获取模型配置失败: %w", err) @@ -43,7 +43,7 @@ func ModelCallResultLambda(ctx context.Context, modelId int64, sessionId string, // 推理模型：分批调用结果需拼接为单个字段，模型类型仅网关配置携带，此处顺带判断 isInference := modelInfo.ModelManage.ModelType != nil && *modelInfo.ModelManage.ModelType == model.TypeInference // 统一异步入口：提交落库 flow_async_task，崩溃恢复重订阅 msg_topic 拿回结果（同步模型直接调用，不落库） - responseParams, err := AsyncModelCallWithRecovery(ctx, execId, nodeId, segIdx, modelId, modelInfo.ModelManage.ResponseType, sessionId, modelRequestParams, businessParams) + responseParams, err := AsyncModelCallWithRecovery(ctx, nodeGroupId, execId, nodeId, segIdx, modelId, modelInfo.ModelManage.ResponseType, sessionId, modelRequestParams, businessParams) if err != nil { return nil, nil, false, err }