Task re execute

[iwanghc]

User description

关联的 issue

#3145

描述你的变更

上线失败SQL重新执行功能实现

确认项（pr提交后操作）

Tip

请在指定复审人之前，确认并完成以下事项，完成后✅

---

• 我已完成自测
• 我已记录完整日志方便进行诊断
• 我已在关联的issue里补充了实现方案
• 我已在关联的issue里补充了测试影响面
• 我已确认了变更的兼容性，如果不兼容则在issue里标记 not_compatible
• 我已确认了是否要更新文档，如果要更新则在issue里标记 need_update_doc

---

---

Description

• 新增任务失败SQL重新执行入口

• 添加任务及工作流状态校验逻辑

• 修改任务执行状态判断和数据库查询接口

• 补充execute_sql_detail查询相关测试

---

Diagram Walkthrough

flowchart LR
  A["API Controller"]
  B["PrepareForTaskReExecution"]
  C["Storage更新与任务查询"]
  D["异步执行ReExecuteTaskSQLs"]
  E["通知工作流执行结果"]
  A -- "调用校验" --> B
  B -- "验证任务/SQL状态" --> C
  C -- "更新工作流状态" --> D
  D -- "反馈结果" --> E

Loading

File Walkthrough

	Relevant files
Enhancement	workflow.go 新增任务重新执行接口及校验逻辑                                                                                    sqle/api/controller/v1/workflow.go 增加获取项目UID及任务详情 新增PrepareForTaskReExecution函数 校验workflow和task错误状态 调用ReExecuteTaskSQLs执行SQL重试 +92/-0    task.go 更新任务模型及支持部分SQL重新执行                                                                              sqle/model/task.go 修改任务执行状态判断条件 添加GetTaskDetailByIdWithExecSqlIds函数 新增GetExecSqlsByTaskIdAndStatus函数 +33/-1    workflow.go 新增工作流记录更新函数用于SQL重执行                                                                            sqle/model/workflow.go 新增UpdateWorkflowExecInstanceRecordForReExecute函数 添加辅助更新工作流记录函数 +21/-0    sqled.go 修改任务添加流程支持部分SQL重执行                                                                              sqle/server/sqled.go 修改addTask接口增加execSqlIds参数 添加AddTaskWaitResultWithSQLIds新接口 更新任务执行状态判断逻辑 +20/-10  workflow.go 新增工作流重新执行任务SQL逻辑                                                                                  sqle/server/workflow.go 新增ReExecuteTaskSQLs函数处理SQL重新执行 更新工作流状态和记录执行人信息 异步执行任务并发送通知 +75/-0
Tests	sqled_test.go 补充execute_sql_detail查询测试                                                                  sqle/server/sqled_test.go 新增对execute_sql_detail查询的mock测试 +2/-0

---

[github-actions]

PR Reviewer Guide 🔍

(Review updated until commit 29e7c64)

🎫 Ticket compliance analysis 🔶 3145 - Partially compliant Compliant requirements: 新增重新执行入口及接口实现 实现工单和任务状态的校验逻辑 补充了针对execute_sql_detail查询和部分异步流程的测试 Non-compliant requirements: [] Requires further human verification: []

⏱️ Estimated effort to review: 3 🔵🔵🔵⚪⚪

🧪 PR contains tests

🔒 No security concerns identified

⚡ Recommended focus areas for review 文案优化 建议修改权限校验失败时返回的错误提示文案，将 "you are not allow to execute the task" 修改为符合英语语法的表达，以提升用户体验 return e.New("you are not allow to execute the task") 逻辑验证 请确认 HasDoingExecute 方法更新后的条件判断逻辑是否完全符合预期业务规则，避免遗漏任务状态判断或出现执行流程异常 if commitSQL.ExecStatus != SQLExecuteStatusInitialized && commitSQL.ExecStatus != SQLExecuteStatusFailed { return true } } 参数传递 检查新增的 execSqlIds 参数在函数 addTask 内的使用，确保在通过 GetTaskDetailByIdWithExecSqlIds 获取任务详情时与历史逻辑保持一致 task, exist, err := st.GetTaskDetailByIdWithExecSqlIds(taskId, execSqlIds) if err != nil {

[github-actions]

Failed to generate code suggestions for PR

[github-actions]

Persistent review updated to latest commit c80ccf3

[github-actions]

PR Code Suggestions ✨

Explore these optional code suggestions:

Category	Suggestion	Impact
Possible issue	处理未使用的通道 建议检查并优化 workflowStatusChan 的使用，目前该通道创建后并没有被消费，可能导致更新状态过程中阻塞或资源泄漏。可以考虑移除该通道或确保其正确使用与关闭以防止潜在问题。 sqle/server/workflow.go [57-73] -workflowStatusChan := make(chan string, 1) +// 如果 updateStatus 不需要通过通道同步结果，则可以直接移除 workflowStatusChan var lock sync.Mutex go func() { sqledServer := GetSqled() task, err := sqledServer.AddTaskWaitResultWithSQLIds(string(workflow.ProjectId), strconv.Itoa(int(task.ID)), execSqlIds, ActionTypeExecute) - { // NOTE: Update the workflow status before sending notifications to ensure that the notification content reflects the latest information. - lock.Lock() - updateStatus(s, workflow, l, workflowStatusChan) - lock.Unlock() - } + lock.Lock() + updateStatus(s, workflow, l, nil) + lock.Unlock() if err != nil || task.Status == model.TaskStatusExecuteFailed { go notification.NotifyWorkflow(string(workflow.ProjectId), workflow.WorkflowId, notification.WorkflowNotifyTypeExecuteFail) } else { go notification.NotifyWorkflow(string(workflow.ProjectId), workflow.WorkflowId, notification.WorkflowNotifyTypeExecuteSuccess) } - }() Suggestion importance[1-10]: 7 __ Why: The suggestion points out that the workflowStatusChan is created but never consumed, which can lead to potential blocking or resource leakage; removing or properly handling it improves reliability.	Medium
General	移除 goto 控制流程 建议移除 goto 语句，通过使用标志变量或者提前返回来简化控制流程，这样代码更易读且更易维护，避免潜在的流程跳转错误。 sqle/api/controller/v1/workflow.go [1424-1438] +allowed := false for _, record := range workflow.Record.InstanceRecords { - if record.TaskId != task.ID { - continue - } - - for _, u := range strings.Split(record.ExecutionAssignees, ",") { - if u == user.GetIDStr() { - goto CheckReExecSqlIds - } + if record.TaskId != task.ID { + continue + } + for _, u := range strings.Split(record.ExecutionAssignees, ",") { + if u == user.GetIDStr() { + allowed = true + break } } + if allowed { + break + } +} +if !allowed { return e.New("you are not allow to execute the task") +} -CheckReExecSqlIds: +// 校验 reExecSqlIds Suggestion importance[1-10]: 6 __ Why: The suggestion replaces the goto statement with a flag variable to simplify the control flow, enhancing readability and maintainability without altering the core functionality.	Low

[github-actions]

Persistent review updated to latest commit f7178ef

[github-actions]

Failed to generate code suggestions for PR

[github-actions]

Persistent review updated to latest commit cd39799

[github-actions]

PR Code Suggestions ✨

Explore these optional code suggestions:

Category	Suggestion	Impact
Possible issue	处理未使用的通道 建议检查并优化 workflowStatusChan 的使用，目前该通道创建后并没有被消费，可能导致更新状态过程中阻塞或资源泄漏。可以考虑移除该通道或确保其正确使用与关闭以防止潜在问题。 sqle/server/workflow.go [57-73] -workflowStatusChan := make(chan string, 1) +// 如果 updateStatus 不需要通过通道同步结果，则可以直接移除 workflowStatusChan var lock sync.Mutex go func() { sqledServer := GetSqled() task, err := sqledServer.AddTaskWaitResultWithSQLIds(string(workflow.ProjectId), strconv.Itoa(int(task.ID)), execSqlIds, ActionTypeExecute) - { // NOTE: Update the workflow status before sending notifications to ensure that the notification content reflects the latest information. - lock.Lock() - updateStatus(s, workflow, l, workflowStatusChan) - lock.Unlock() - } + lock.Lock() + updateStatus(s, workflow, l, nil) + lock.Unlock() if err != nil || task.Status == model.TaskStatusExecuteFailed { go notification.NotifyWorkflow(string(workflow.ProjectId), workflow.WorkflowId, notification.WorkflowNotifyTypeExecuteFail) } else { go notification.NotifyWorkflow(string(workflow.ProjectId), workflow.WorkflowId, notification.WorkflowNotifyTypeExecuteSuccess) } - }() Suggestion importance[1-10]: 7 __ Why: The suggestion points out that the workflowStatusChan is created but never consumed, which can lead to potential blocking or resource leakage; removing or properly handling it improves reliability.	Medium
General	移除 goto 控制流程 建议移除 goto 语句，通过使用标志变量或者提前返回来简化控制流程，这样代码更易读且更易维护，避免潜在的流程跳转错误。 sqle/api/controller/v1/workflow.go [1424-1438] +allowed := false for _, record := range workflow.Record.InstanceRecords { - if record.TaskId != task.ID { - continue - } - - for _, u := range strings.Split(record.ExecutionAssignees, ",") { - if u == user.GetIDStr() { - goto CheckReExecSqlIds - } + if record.TaskId != task.ID { + continue + } + for _, u := range strings.Split(record.ExecutionAssignees, ",") { + if u == user.GetIDStr() { + allowed = true + break } } + if allowed { + break + } +} +if !allowed { return e.New("you are not allow to execute the task") +} -CheckReExecSqlIds: +// 校验 reExecSqlIds Suggestion importance[1-10]: 6 __ Why: The suggestion replaces the goto statement with a flag variable to simplify the control flow, enhancing readability and maintainability without altering the core functionality.	Low

[github-actions]

Persistent review updated to latest commit 29e7c64

[github-actions]

Failed to generate code suggestions for PR