diff --git a/workflow/dao/flow/flow_segment_result_dao.go b/workflow/dao/flow/flow_segment_result_dao.go index 11fd2ed..175d023 100644 --- a/workflow/dao/flow/flow_segment_result_dao.go +++ b/workflow/dao/flow/flow_segment_result_dao.go @@ -4,6 +4,7 @@ import ( "ai-agent/workflow/consts/public" "ai-agent/workflow/model/entity" "context" + "fmt" "gitea.redpowerfuture.com/red-future/common/db/gfdb" ) @@ -47,11 +48,12 @@ func (d *flowSegmentResultDao) ListByNode(ctx context.Context, execId int64, nod return m, nil } -// DeleteByExecution 清理指定执行的段结果(全新执行前清旧段 / 工作流执行成功后清理) +// 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 { _, err := gfdb.DB(ctx, public.DbNameBlackDeacon). - Model(ctx, public.TableNameFlowSegmentResult). - Where(entity.FlowSegmentResultCol.ExecutionId, execId). - Delete() + Exec(ctx, fmt.Sprintf("DELETE FROM %s WHERE execution_id = ?", public.TableNameFlowSegmentResult), execId) return err } diff --git a/workflow/dao/flow/flow_segment_result_dao_test.go b/workflow/dao/flow/flow_segment_result_dao_test.go index 978df16..16068b1 100644 --- a/workflow/dao/flow/flow_segment_result_dao_test.go +++ b/workflow/dao/flow/flow_segment_result_dao_test.go @@ -63,51 +63,51 @@ func TestFlowSegmentResultDao(t *testing.T) { } } -// TestFlowSegmentResultDaoResurrect 软删→重存的"复活"路径验证。 -// 实体嵌入 beans.SQLBaseDO(含 deleted_at),GoFrame 将其识别为软删除列: -// DeleteByExecution 实际执行的是 UPDATE ... SET deleted_at=now(),而非物理 DELETE。 -// 本测试验证随后对同一 (execution_id, node_id, segment_index) 的 Save(命中 -// ON CONFLICT ... DO UPDATE)是否会清除 deleted_at,使 ListByNode 再次可见该段。 -// 若不可见,说明 upsert 的 SET 子句不含 deleted_at,复活路径被破坏 → 续跑复用静默失效。 -func TestFlowSegmentResultDaoResurrect(t *testing.T) { +// TestFlowSegmentResultDaoReuseAfterCleanup 硬删→重存的"续跑复用"保证。 +// DeleteByExecution 必须物理删除:实体嵌入 beans.SQLBaseDO(含 deleted_at),若走 +// gfdb Model.Delete() 会退化为软删(UPDATE ... SET deleted_at=now()),而后续 Save +// 的 ON CONFLICT ... DO UPDATE SET 不含 deleted_at(OmitNil 丢弃 nil),软删行无法复活, +// ListByNode 永远看不到 → 每次重执行都全量重生成。本测试验证:清理(物理删两行)后 +// 对同一 (execution_id, node_id, segment_index) 重存段2,仅段2可见(len==1)且为更新后 URL。 +func TestFlowSegmentResultDaoReuseAfterCleanup(t *testing.T) { if os.Getenv("AI_AGENT_TEST_DB") == "" { t.Skip("设置 AI_AGENT_TEST_DB=1 且本地 black_deacon 可用时运行") } ctx := context.WithValue(context.Background(), "user", &beans.User{UserName: "unit_test", TenantId: 999999}) execId := time.Now().UnixMilli() - nodeId := "seg_test_resurrect" + nodeId := "seg_test_cleanup" dao := FlowSegmentResultDao - // 软删即可,物理残留由外部硬清理(docker exec psql DELETE FROM)负责 + // 物理残留由硬清理(docker exec psql DELETE FROM)兜底 t.Cleanup(func() { _ = dao.DeleteByExecution(ctx, execId) }) - if err := dao.Save(ctx, execId, nodeId, 1, "video_url", "http://host/u1-resurrect.mp4"); err != nil { + if err := dao.Save(ctx, execId, nodeId, 1, "video_url", "http://host/u1-cleanup.mp4"); err != nil { t.Fatalf("Save 段1失败: %v", err) } - if err := dao.Save(ctx, execId, nodeId, 2, "video_url", "http://host/u2-resurrect.mp4"); err != nil { + if err := dao.Save(ctx, execId, nodeId, 2, "video_url", "http://host/u2-cleanup.mp4"); err != nil { t.Fatalf("Save 段2失败: %v", err) } - // 软删除整条执行 + // 物理删除整条执行(DELETE FROM ... WHERE execution_id=?) if err := dao.DeleteByExecution(ctx, execId); err != nil { t.Fatalf("DeleteByExecution失败: %v", err) } - // 重新保存段2(同一 exec/node/idx,不同 URL): - // 若 ON CONFLICT DO UPDATE 的 SET 子句不含 deleted_at,段2 在 ListByNode 中仍不可见 - if err := dao.Save(ctx, execId, nodeId, 2, "video_url", "http://host/u2-resurrect.mp4"); err != nil { - t.Fatalf("Save 段2(软删后重存)失败: %v", err) + // 清理后重新保存段2(同一 exec/node/idx,不同 URL): + // 物理删除已移除该行,Save 走纯 INSERT,段2 在 ListByNode 中可见且为更新后的 URL + if err := dao.Save(ctx, execId, nodeId, 2, "video_url", "http://host/u2-cleanup-new.mp4"); err != nil { + t.Fatalf("Save 段2(清理后重存)失败: %v", err) } m, err := dao.ListByNode(ctx, execId, nodeId) if err != nil { t.Fatalf("ListByNode失败: %v", err) } - // 期望:段2 复活(deleted_at 被 upsert 清除),len(m)==2 且 m[2] 为更新后的 URL - if len(m) != 2 { - t.Fatalf("软删后重存应复活段2(len=2),实际 len=%d: %+v", len(m), m) + // 期望:段1 已物理删除,仅段2 可见(len==1)且为更新后的 URL + if len(m) != 1 { + t.Fatalf("硬删后重存应只剩段2(len=1),实际 len=%d: %+v", len(m), m) } - if m[2].URL != "http://host/u2-resurrect.mp4" { - t.Fatalf("段2 应恢复为更新后的 URL,实际: %+v", m[2]) + if m[2].URL != "http://host/u2-cleanup-new.mp4" { + t.Fatalf("段2 应为更新后的 URL,实际: %+v", m[2]) } }