19904408334
73f296731c
将 checkpoint、异步任务与段结果缓存的唯一键从 execution_id 改为 node_group_id，区分换参重跑与续跑语义；恢复/续跑复用执行记录的组标识，换参重跑则换新组并软删旧组残留，消除软删墓碑导致同键重存失效的问题。
68 lines
2.9 KiB
Go
68 lines
2.9 KiB
Go
package flow
|
||
|
||
import (
|
||
"ai-agent/workflow/consts/public"
|
||
"ai-agent/workflow/model/entity"
|
||
"context"
|
||
|
||
"gitea.redpowerfuture.com/red-future/common/db/gfdb"
|
||
)
|
||
|
||
var FlowSegmentResultDao = &flowSegmentResultDao{}
|
||
|
||
type flowSegmentResultDao struct{}
|
||
|
||
// 缓存唯一键是 (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,
|
||
VideoKey: videoKey,
|
||
VideoURL: videoURL,
|
||
}
|
||
_, err := gfdb.DB(ctx, public.DbNameBlackDeacon).
|
||
Model(ctx, public.TableNameFlowSegmentResult).
|
||
Data(rec).
|
||
OnConflict(entity.FlowSegmentResultCol.NodeGroupId, entity.FlowSegmentResultCol.NodeId, entity.FlowSegmentResultCol.SegmentIndex).
|
||
Save()
|
||
return err
|
||
}
|
||
|
||
// 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.NodeGroupId, nodeGroupId).
|
||
Where(entity.FlowSegmentResultCol.NodeId, nodeId).
|
||
Scan(&list)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
m := make(map[int]entity.SegmentRef, len(list))
|
||
for _, r := range list {
|
||
m[r.SegmentIndex] = entity.SegmentRef{Key: r.VideoKey, URL: r.VideoURL}
|
||
}
|
||
return m, nil
|
||
}
|
||
|
||
// 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).
|
||
Model(ctx, public.TableNameFlowSegmentResult).
|
||
Where(entity.FlowSegmentResultCol.NodeGroupId, nodeGroupId).
|
||
Delete()
|
||
return err
|
||
}
|