122 lines
4.4 KiB
Go
122 lines
4.4 KiB
Go
package repository
|
|
|
|
import (
|
|
"context"
|
|
"database/sql"
|
|
"errors"
|
|
"fmt"
|
|
"strings"
|
|
"time"
|
|
|
|
"git.pcux.cn/xuans/Scert/backend/internal/deployment"
|
|
)
|
|
|
|
// TaskCompletion 同时关闭任务与对应尝试，避免只保存其中一方的终态。
|
|
type TaskCompletion struct {
|
|
TaskID string
|
|
AttemptID string
|
|
Status string
|
|
Stage string
|
|
Message string
|
|
StartedAt *time.Time
|
|
FinishedAt time.Time
|
|
Report *deployment.Report
|
|
AuditAction string
|
|
Retry bool
|
|
WithoutAttempt bool
|
|
}
|
|
|
|
func (r *SQLTaskRepository) FinishTaskAttempt(ctx context.Context, completion TaskCompletion) error {
|
|
if r == nil || r.db == nil {
|
|
return errors.New("数据库连接不能为空")
|
|
}
|
|
completion.TaskID = strings.TrimSpace(completion.TaskID)
|
|
completion.AttemptID = strings.TrimSpace(completion.AttemptID)
|
|
completion.Stage = strings.TrimSpace(completion.Stage)
|
|
if completion.TaskID == "" || (completion.AttemptID == "" && !completion.WithoutAttempt) || completion.Stage == "" || completion.FinishedAt.IsZero() {
|
|
return errors.New("任务终态记录不完整")
|
|
}
|
|
switch completion.Status {
|
|
case "success", "failed", "cancelled", "partial_success":
|
|
default:
|
|
return errors.New("任务终态无效")
|
|
}
|
|
if completion.WithoutAttempt && (completion.AttemptID != "" || completion.Report != nil || completion.Retry || (completion.Status != "failed" && completion.Status != "cancelled")) {
|
|
return errors.New("无活动尝试的收尾参数无效")
|
|
}
|
|
if completion.Retry && (completion.Status != "failed" || completion.AuditAction != "") {
|
|
return errors.New("重试只能保存失败尝试且不能发布完成审计")
|
|
}
|
|
tx, err := r.db.BeginTx(ctx, nil)
|
|
if err != nil {
|
|
return fmt.Errorf("开始任务收尾事务失败：%w", err)
|
|
}
|
|
defer tx.Rollback()
|
|
taskStatus, taskStage := completion.Status, completion.Stage
|
|
var taskFinished any = completion.FinishedAt.UTC()
|
|
activeStates := "('queued', 'running', 'retry_wait', 'cancelling')"
|
|
if completion.Retry {
|
|
taskStatus, taskStage, taskFinished = "retry_wait", "retry_wait", nil
|
|
// 已受理的取消不能被重试重新激活。
|
|
activeStates = "('running')"
|
|
}
|
|
// 与取消、重试保持任务在前、尝试在后的加锁顺序。旧尝试或跨任务尝试必须整体回滚。
|
|
result, err := tx.ExecContext(ctx, r.query(`UPDATE tasks
|
|
SET status = ?, stage = ?, message = ?, started_at = COALESCE(?, started_at), finished_at = ?, updated_at = ?
|
|
WHERE id = ? AND status IN `+activeStates),
|
|
taskStatus, taskStage, completion.Message, nullableTime(completion.StartedAt),
|
|
taskFinished, completion.FinishedAt.UTC(), completion.TaskID)
|
|
if err != nil {
|
|
return fmt.Errorf("保存任务终态失败：%w", err)
|
|
}
|
|
if count, err := result.RowsAffected(); err != nil {
|
|
return fmt.Errorf("读取任务收尾结果失败：%w", err)
|
|
} else if count != 1 {
|
|
return sql.ErrNoRows
|
|
}
|
|
if err := r.finishCompletionAttempt(ctx, tx, completion); err != nil {
|
|
return err
|
|
}
|
|
if completion.Report != nil {
|
|
if err := r.recordCompletionReport(ctx, tx, completion); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
if completion.AuditAction != "" {
|
|
if err := r.recordCompletionAudit(ctx, tx, completion); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
if err := tx.Commit(); err != nil {
|
|
return fmt.Errorf("提交任务收尾事务失败：%w", err)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (r *SQLTaskRepository) finishCompletionAttempt(ctx context.Context, tx *sql.Tx, completion TaskCompletion) error {
|
|
if completion.WithoutAttempt {
|
|
var active int
|
|
if err := tx.QueryRowContext(ctx, r.query(`SELECT COUNT(*) FROM task_attempts
|
|
WHERE task_id = ? AND status IN ('queued', 'running', 'retry_wait', 'cancelling')`), completion.TaskID).Scan(&active); err != nil {
|
|
return fmt.Errorf("检查活动尝试失败：%w", err)
|
|
}
|
|
if active != 0 {
|
|
return errors.New("任务仍有活动尝试，不能单独收尾")
|
|
}
|
|
return nil
|
|
}
|
|
result, err := tx.ExecContext(ctx, r.query(`UPDATE task_attempts
|
|
SET status = ?, stage = ?, message = ?, finished_at = ?
|
|
WHERE id = ? AND task_id = ? AND status IN ('queued', 'running', 'retry_wait', 'cancelling')`),
|
|
completion.Status, completion.Stage, completion.Message, completion.FinishedAt.UTC(), completion.AttemptID, completion.TaskID)
|
|
if err != nil {
|
|
return fmt.Errorf("保存任务尝试终态失败：%w", err)
|
|
}
|
|
if count, err := result.RowsAffected(); err != nil {
|
|
return fmt.Errorf("读取任务尝试收尾结果失败：%w", err)
|
|
} else if count != 1 {
|
|
return sql.ErrNoRows
|
|
}
|
|
return nil
|
|
}
|