diff --git a/update.sql b/update.sql index 46b2c08..d0d64d4 100644 --- a/update.sql +++ b/update.sql @@ -887,3 +887,35 @@ COMMENT ON COLUMN black_deacon_flow_segment_result.segment_index IS '段序号'; COMMENT ON COLUMN black_deacon_flow_segment_result.video_key IS '视频输出字段key'; COMMENT ON COLUMN black_deacon_flow_segment_result.video_url IS '已生成成功的视频地址'; --------------------pgsql创建black_deacon_flow_segment_result表语句--------------------------- + +--------------------工作流重试兜底：exec_workflow 新增重试/心跳列--------------------- +ALTER TABLE black_deacon_exec_workflow ADD COLUMN IF NOT EXISTS retryable SMALLINT NOT NULL DEFAULT 0; -- 1=程序报错可重试,0=用户取消 +ALTER TABLE black_deacon_exec_workflow ADD COLUMN IF NOT EXISTS retry_count INTEGER NOT NULL DEFAULT 0; -- 已重试次数 +ALTER TABLE black_deacon_exec_workflow ADD COLUMN IF NOT EXISTS last_heartbeat BIGINT NOT NULL DEFAULT 0; -- 最后心跳（毫秒时间戳） +COMMENT ON COLUMN black_deacon_exec_workflow.retryable IS '是否可重试:0-用户取消,1-程序报错'; +COMMENT ON COLUMN black_deacon_exec_workflow.retry_count IS '已重试次数'; +COMMENT ON COLUMN black_deacon_exec_workflow.last_heartbeat IS '最后心跳时间（毫秒时间戳）'; + +--------------------工作流重试兜底：flow_async_task 统一异步任务表（Task 2 使用）--------------------- +CREATE TABLE IF NOT EXISTS black_deacon_flow_async_task ( + -- 基础字段（完全对齐项目规范） + id BIGINT PRIMARY KEY, + tenant_id BIGINT NOT NULL DEFAULT 0, + creator VARCHAR(64) NOT NULL, + created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP, + updater VARCHAR(64) NOT NULL, + updated_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP, + deleted_at timestamp(6), + -- 业务字段 + execution_id BIGINT NOT NULL DEFAULT 0, -- 所属执行ID + node_id VARCHAR(64) NOT NULL DEFAULT '', -- 所属节点ID + segment_index INT NOT NULL DEFAULT -1, -- 段序号；非段调用为-1（哨兵，不用NULL） + model_id BIGINT NOT NULL DEFAULT 0, -- 模型ID + task_id BIGINT NOT NULL DEFAULT 0, -- model-gateway任务ID + msg_topic VARCHAR(255) NOT NULL DEFAULT '', -- 结果消息主题（恢复重订阅拿回结果） + state SMALLINT NOT NULL DEFAULT 0, -- 0=in-flight,1=done,2=failed + result JSONB DEFAULT '{}' -- 成功结果JSON(ModelCallRes) +); +CREATE UNIQUE INDEX IF NOT EXISTS uk_async_task_exec_node_seg ON black_deacon_flow_async_task(execution_id, node_id, segment_index); +CREATE INDEX IF NOT EXISTS idx_async_task_tenant_id ON black_deacon_flow_async_task(tenant_id); +CREATE INDEX IF NOT EXISTS idx_async_task_deleted_at ON black_deacon_flow_async_task(deleted_at); diff --git a/workflow/dao/session/exec_workflow_dao.go b/workflow/dao/session/exec_workflow_dao.go index 6043a05..c7144d6 100644 --- a/workflow/dao/session/exec_workflow_dao.go +++ b/workflow/dao/session/exec_workflow_dao.go @@ -1,10 +1,13 @@ package session import ( + flow "ai-agent/workflow/consts/flow" "ai-agent/workflow/consts/public" sessionDto "ai-agent/workflow/model/dto/session" "ai-agent/workflow/model/entity" "context" + "fmt" + "time" "gitea.redpowerfuture.com/red-future/common/beans" "gitea.redpowerfuture.com/red-future/common/db/gfdb" @@ -117,3 +120,55 @@ func (d *execWorkflowDao) ListBySession(ctx context.Context, sessionId string) ( err = r.Structs(&res) return } + +// ResetRunning 置为运行中并刷新心跳（execute 复用失败记录 / reExecute / 恢复例程共用； +// 用 map 更新避免 OmitEmpty 省略 0 值，同时写 status、last_heartbeat、node_group_id） +func (d *execWorkflowDao) ResetRunning(ctx context.Context, id int64, nodeGroupId string) error { + _, err := gfdb.DB(ctx, public.DbNameBlackDeacon).Model(ctx, public.TableNameExecWorkflow). + Where(entity.ExecWorkflowCol.Id, id). + Data(map[string]any{ + entity.ExecWorkflowCol.Status: gconv.Int8(*flow.FlowExecutionStatusRunning.Code()), + entity.ExecWorkflowCol.LastHeartbeat: time.Now().UnixMilli(), + entity.ExecWorkflowCol.NodeGroupId: nodeGroupId, + }). + Update() + return err +} + +// TouchHeartbeat 更新执行心跳（毫秒时间戳），供后台心跳 goroutine 每 30s 调用一次 +func (d *execWorkflowDao) TouchHeartbeat(ctx context.Context, id int64) error { + _, err := gfdb.DB(ctx, public.DbNameBlackDeacon).Model(ctx, public.TableNameExecWorkflow). + Where(entity.ExecWorkflowCol.Id, id). + Data(map[string]any{entity.ExecWorkflowCol.LastHeartbeat: time.Now().UnixMilli()}). + Update() + return err +} + +// UpdateRetry 更新重试标记与已重试次数（map 更新，retryable=0 也需写入） +func (d *execWorkflowDao) UpdateRetry(ctx context.Context, id int64, retryable int, retryCount int) error { + _, err := gfdb.DB(ctx, public.DbNameBlackDeacon).Model(ctx, public.TableNameExecWorkflow). + Where(entity.ExecWorkflowCol.Id, id). + Data(map[string]any{ + entity.ExecWorkflowCol.Retryable: retryable, + entity.ExecWorkflowCol.RetryCount: retryCount, + }). + Update() + return err +} + +// ListRecoverable 返回可恢复执行：僵尸运行中（status=1 且心跳陈旧）或可重试失败（status=3 且 retryable=1 且未耗尽） +func (d *execWorkflowDao) ListRecoverable(ctx context.Context, now int64, staleBefore int64, maxRetry int) (res []*entity.ExecWorkflow, err error) { + r, err := gfdb.DB(ctx, public.DbNameBlackDeacon).Model(ctx, public.TableNameExecWorkflow). + Where(fmt.Sprintf("(%s = ? AND %s < ?) OR (%s = ? AND %s = 1 AND %s < ?)", + entity.ExecWorkflowCol.Status, entity.ExecWorkflowCol.LastHeartbeat, + entity.ExecWorkflowCol.Status, entity.ExecWorkflowCol.Retryable, entity.ExecWorkflowCol.RetryCount), + gconv.Int8(*flow.FlowExecutionStatusRunning.Code()), staleBefore, + gconv.Int8(*flow.FlowExecutionStatusFailed.Code()), maxRetry). + All() + if err != nil { + return + } + // 用 All+Structs 而非 Scan：Scan 生成的列清单会丢嵌入 SQLBaseDO 的 id 等基础列，恢复例程需要 id + err = r.Structs(&res) + return +} diff --git a/workflow/dao/session/exec_workflow_dao_test.go b/workflow/dao/session/exec_workflow_dao_test.go new file mode 100644 index 0000000..b845511 --- /dev/null +++ b/workflow/dao/session/exec_workflow_dao_test.go @@ -0,0 +1,106 @@ +package session + +import ( + "context" + "os" + "testing" + "time" + + "ai-agent/workflow/consts/flow" + sessionDto "ai-agent/workflow/model/dto/session" + "ai-agent/workflow/model/entity" + + "gitea.redpowerfuture.com/red-future/common/beans" + _ "github.com/gogf/gf/contrib/drivers/pgsql/v2" +) + +// TestExecWorkflowRetryColsRoundTrip 验证三新列读写 + ListRecoverable 三类判定（需 AI_AGENT_TEST_DB=1 且本地 PG 可用） +func TestExecWorkflowRetryColsRoundTrip(t *testing.T) { + if !isTestDBEnabled(t) { + t.Skip("skip: AI_AGENT_TEST_DB not set") + } + // gfdb 的 Hook 依赖 ctx 中注入用户信息（租户/创建人），否则 Insert 报 "token 数据为空" + ctx := context.WithValue(context.Background(), "user", &beans.User{ + UserName: "test-retry", TenantId: 1, + }) + id, err := ExecWorkflowDao.Insert(ctx, &sessionDto.CreateWorkflowReq{ + SessionId: "test-retry", FlowId: 1, NodeGroupId: "ng", + Status: flow.FlowExecutionStatusRunning.Code(), + LastHeartbeat: time.Now().UnixMilli(), + }) + if err != nil { + t.Fatalf("insert: %v", err) + } + defer ExecWorkflowDao.Delete(ctx, &sessionDto.DeleteExecWorkflowReq{Id: []int64{id}}) + + if err := ExecWorkflowDao.TouchHeartbeat(ctx, id); err != nil { + t.Fatalf("touch heartbeat: %v", err) + } + if err := ExecWorkflowDao.UpdateRetry(ctx, id, 1, 1); err != nil { + t.Fatalf("update retry: %v", err) + } + got, err := ExecWorkflowDao.GetById(ctx, id) + if err != nil || got == nil { + t.Fatalf("get: %v", err) + } + if got.Retryable != 1 || got.RetryCount != 1 || got.LastHeartbeat == 0 { + t.Fatalf("cols not persisted: retryable=%d retryCount=%d heartbeat=%d", got.Retryable, got.RetryCount, got.LastHeartbeat) + } + + now := time.Now().UnixMilli() + stale := now - int64(120*time.Second/time.Millisecond) + if err := ExecWorkflowDao.UpdateRetry(ctx, id, 0, 0); err != nil { + t.Fatalf("update retry reset: %v", err) + } + // 情况1：status=3 retryable=1 retry_count